| 308 | __slots__ = ('_r', '_w', '_on_end', '_s1', '_s2', '_t1', '_t2') |
| 309 | |
| 310 | def __init__(self, r, w, on_end=None): |
| 311 | self._r = r |
| 312 | self._w = w |
| 313 | self._on_end = on_end |
| 314 | |
| 315 | self._s1, self._s2 = socket.socketpair() |
| 316 | debug3("[SocketShim] r=%r w=%r | s1=%r s2=%r" % (self._r, self._w, self._s1, self._s2)) |
| 317 | |
| 318 | def stream_reader_to_sock(): |
| 319 | try: |
| 320 | for data in iter(lambda: self._r.read(16384), b''): |
| 321 | self._s1.sendall(data) |
| 322 | # debug3("[SocketRWShim] <<<<< r.read() %d %r..." % (len(data), data[:min(32, len(data))])) |
| 323 | except Exception: |
| 324 | traceback.print_exc(file=sys.stderr) |
| 325 | finally: |
| 326 | debug2("[SocketRWShim] Thread 'stream_reader_to_sock' exiting") |
| 327 | self._s1.close() |
| 328 | self._on_end and self._on_end() |
| 329 | |
| 330 | def stream_sock_to_writer(): |
| 331 | try: |
| 332 | for data in iter(lambda: self._s1.recv(16384), b''): |
| 333 | while data: |
| 334 | n = self._w.write(data) |
| 335 | data = data[n:] |
| 336 | # debug3("[SocketRWShim] <<<<< w.write() %d %r..." % (len(data), data[:min(32, len(data))])) |
| 337 | except Exception: |
| 338 | traceback.print_exc(file=sys.stderr) |
| 339 | finally: |
| 340 | debug2("[SocketRWShim] Thread 'stream_sock_to_writer' exiting") |
| 341 | self._s1.close() |
| 342 | self._on_end and self._on_end() |
| 343 | |
| 344 | self._t1 = threading.Thread(target=stream_reader_to_sock, name='stream_reader_to_sock', daemon=True).start() |
| 345 | self._t2 = threading.Thread(target=stream_sock_to_writer, name='stream_sock_to_writer', daemon=True).start() |
| 346 | |
| 347 | def makefiles(self): |
| 348 | return self._s2.makefile("rb", buffering=0), self._s2.makefile("wb", buffering=0) |