I came up with the following method to pass bytes to Popen in a non-blocking way.
Is there a better way or a suitable library?
import os
from threading import Thread
from subprocess import Popen
from time import sleep
def get_read_stream(data: bytes):
r, w = os.pipe()
def _write():
with open(w, 'wb') as f:
f.write(data)
sleep(5) # <- To check for asynchronicity
print('closed')
Thread(target=_write).start()
return open(r, 'rb')
rs = get_read_stream(b'foobar\n')
proc = Popen(['cat'], stdin=rs)
print('start proc')
proc.wait()
print('end')
P.S.
I've come up with another method. I think this is better because there is less stream to manage.
from threading import Thread
from subprocess import Popen
from time import sleep
def write_to_stream(stream, data: bytes):
try:
stream.write(data)
sleep(5) # <- To check for asynchronicity
finally:
stream.close()
print('closed')
proc = Popen(['cat'], stdin=PIPE)
print('start proc')
Thread(target=write_to_stream, args=(proc.stdin, b'foobar\n')).start()
proc.wait()
print('end')