104 lines
2.6 KiB
Python
104 lines
2.6 KiB
Python
import asyncio
|
|
import nest_asyncio
|
|
|
|
id_counter = 0
|
|
|
|
def run_fn (fn_obj, arg):
|
|
if fn_obj == None:
|
|
return
|
|
|
|
# https://stackoverflow.com/questions/36076619/test-if-function-or-method-is-normal-or-asynchronous
|
|
if asyncio.iscoroutinefunction(fn_obj):
|
|
if arg == None:
|
|
asyncio.create_task(fn_obj())
|
|
else:
|
|
asyncio.create_task(fn_obj(arg))
|
|
return
|
|
|
|
if arg == None:
|
|
fn_obj()
|
|
else:
|
|
fn_obj(arg)
|
|
|
|
def make_runnable(fn_obj):
|
|
"""
|
|
Creates an async signal and a task that listen to the signal.
|
|
When it is received, fn_obj is called.
|
|
The task doesn't stop and keeps forwarding signals to the passed function.
|
|
There can also be arguments coming along with the signal, those are passed to the function.
|
|
"""
|
|
signal_queue = asyncio.Queue()
|
|
|
|
async def signal_listener():
|
|
while True:
|
|
args = await signal_queue.get()
|
|
if args is None: # Stop signal
|
|
break
|
|
run_fn(fn_obj, args[0] if args else None)
|
|
|
|
# Start the listener task in the background
|
|
asyncio.create_task(signal_listener())
|
|
|
|
# Return a method to put items on the signal queue
|
|
return lambda *args: signal_queue.put_nowait(args)
|
|
|
|
class Stream:
|
|
def __init__(self, app):
|
|
global id_counter
|
|
|
|
self.id = id_counter
|
|
id_counter = id_counter + 1
|
|
|
|
self.open = True
|
|
self.content = ""
|
|
self.listeners = []
|
|
|
|
app.streams[self.id] = self
|
|
|
|
def listen (self, chunk_fn, end_fn):
|
|
nest_asyncio.apply()
|
|
#chunk_fn = make_runnable(chunk_fn)
|
|
#end_fn = make_runnable(end_fn)
|
|
|
|
if self.open:
|
|
self.listeners.append({
|
|
"chunk": chunk_fn,
|
|
"end": end_fn,
|
|
})
|
|
|
|
if self.content != "":
|
|
run_fn(chunk_fn, self.content)
|
|
else:
|
|
run_fn(chunk_fn, self.content)
|
|
run_fn(end_fn, None)
|
|
|
|
def feed (self, chunk):
|
|
if not self.open:
|
|
return
|
|
|
|
self.content = self.content + chunk
|
|
for listener in self.listeners:
|
|
run_fn(listener["chunk"], chunk)
|
|
|
|
def close (self):
|
|
self.open = False
|
|
|
|
for listener in self.listeners:
|
|
run_fn(listener["end"], None)
|
|
|
|
async def content_wait (self):
|
|
async_loop = asyncio.get_event_loop()
|
|
event = asyncio.Event()
|
|
|
|
def set_event():
|
|
event.set()
|
|
|
|
def noop (chunk): pass
|
|
def end_fn ():
|
|
async_loop.call_soon_threadsafe(set_event)
|
|
|
|
self.listen(chunk_fn=noop, end_fn=end_fn)
|
|
await event.wait()
|
|
|
|
return self.content
|