import io
import os
import sys
-import pickle
import select
import socket
import struct
"""Send a (picklable) object"""
self._check_closed()
self._check_writable()
- buf = io.BytesIO()
- ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(obj)
- self._send_bytes(buf.getbuffer())
+ self._send_bytes(ForkingPickler.dumps(obj))
def recv_bytes(self, maxlength=None):
"""
self._check_closed()
self._check_readable()
buf = self._recv_bytes()
- return pickle.loads(buf.getbuffer())
+ return ForkingPickler.loads(buf.getbuffer())
def poll(self, timeout=0.0):
"""Whether there is any input available to be read"""
# Licensed to PSF under a Contributor Agreement.
#
+import io
import os
+import pickle
import sys
import signal
import errno
def register(cls, type, reduce):
cls._extra_reducers[type] = reduce
+ @staticmethod
+ def dumps(obj):
+ buf = io.BytesIO()
+ ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(obj)
+ return buf.getbuffer()
+
+ loads = pickle.loads
+
+
def _reduce_method(m):
if m.__self__ is None:
return getattr, (m.__class__, m.__func__.__name__)