core.actor: start with simple actor implementation
This commit is contained in:
parent
f0745f4a7b
commit
5b54e76327
6 changed files with 81 additions and 41 deletions
|
|
@ -2,48 +2,22 @@
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
||||||
QUIT = object()
|
from scopes import util
|
||||||
|
|
||||||
|
quit = util.Symbol("quit")
|
||||||
|
|
||||||
async def loop(q, bhv):
|
async def loop(q, bhv):
|
||||||
while True:
|
while True:
|
||||||
msg = await q.get()
|
msg = await q.get()
|
||||||
if msg == QUIT:
|
if msg is quit:
|
||||||
break
|
break
|
||||||
bhv = await bhv(msg) or bhv
|
bhv = await bhv(msg) or bhv
|
||||||
|
|
||||||
def create(bhv):
|
def create(bhv, group=None):
|
||||||
q = asyncio.Queue()
|
q = asyncio.Queue()
|
||||||
task = asyncio.create_task(loop(q, bhv))
|
if group is None:
|
||||||
return q, task
|
return q, asyncio.create_task(loop(q, bhv))
|
||||||
|
else:
|
||||||
# interactive tests
|
group.create_task(loop(q, bhv))
|
||||||
|
return q
|
||||||
async def handle_basic(msg):
|
|
||||||
match msg[0]:
|
|
||||||
case "show": show(msg)
|
|
||||||
case "switch": return handle_alt
|
|
||||||
case _: show(msg)
|
|
||||||
|
|
||||||
async def handle_alt(msg):
|
|
||||||
await asyncio.sleep(0.01)
|
|
||||||
match msg[0]:
|
|
||||||
case "show": print('no way')
|
|
||||||
case _: show(msg)
|
|
||||||
|
|
||||||
def show(msg):
|
|
||||||
print(msg)
|
|
||||||
|
|
||||||
async def main():
|
|
||||||
q, task = create(handle_basic)
|
|
||||||
await q.put(["Hello World!"])
|
|
||||||
await q.put(["switch"])
|
|
||||||
await q.put(["show", "nope"])
|
|
||||||
await q.put(["Hello World!"])
|
|
||||||
await q.put(QUIT)
|
|
||||||
#await task
|
|
||||||
await asyncio.wait([task])
|
|
||||||
#await asyncio.sleep(10)
|
|
||||||
|
|
||||||
def run():
|
|
||||||
asyncio.run(main())
|
|
||||||
|
|
||||||
|
|
|
||||||
21
scopes/shape.py
Normal file
21
scopes/shape.py
Normal file
|
|
@ -0,0 +1,21 @@
|
||||||
|
# scopes.shape
|
||||||
|
|
||||||
|
"""really abstract basic data shape definitions"""
|
||||||
|
|
||||||
|
class Record:
|
||||||
|
|
||||||
|
headFields = [] # to be defined by subclass
|
||||||
|
|
||||||
|
def __init__(self, *headValues, data=None, **kw):
|
||||||
|
self.head = {}
|
||||||
|
for k, v in kw.items():
|
||||||
|
if k in self.headFields:
|
||||||
|
self.head[k] = kw.pop(k)
|
||||||
|
for ix, hv in enumerate(headValues):
|
||||||
|
self.head[self.headFields[ix]] = hv
|
||||||
|
for k in self.headFields:
|
||||||
|
if self.head.get(k) is None:
|
||||||
|
self.head[k] = ''
|
||||||
|
setattr(self, k, self.head[k])
|
||||||
|
self.data = data or {}
|
||||||
|
|
||||||
|
|
@ -20,14 +20,14 @@ class Track:
|
||||||
headFields = ['taskId', 'userName']
|
headFields = ['taskId', 'userName']
|
||||||
prefix = 'rec'
|
prefix = 'rec'
|
||||||
|
|
||||||
def __init__(self, *keys, data=None, timeStamp=None, trackId=None,
|
def __init__(self, *headValues, data=None, timeStamp=None, trackId=None,
|
||||||
container=None, **kw):
|
container=None, **kw):
|
||||||
self.head = {}
|
self.head = {}
|
||||||
for k, v in kw.items():
|
for k, v in kw.items():
|
||||||
if k in self.headFields:
|
if k in self.headFields:
|
||||||
self.head[k] = kw.pop(k)
|
self.head[k] = kw.pop(k)
|
||||||
for ix, k in enumerate(keys):
|
for ix, hv in enumerate(headValues):
|
||||||
self.head[self.headFields[ix]] = k
|
self.head[self.headFields[ix]] = hv
|
||||||
for k in self.headFields:
|
for k in self.headFields:
|
||||||
if self.head.get(k) is None:
|
if self.head.get(k) is None:
|
||||||
self.head[k] = ''
|
self.head[k] = ''
|
||||||
|
|
|
||||||
45
tests/test_actor.py
Normal file
45
tests/test_actor.py
Normal file
|
|
@ -0,0 +1,45 @@
|
||||||
|
# py-scopes: tests.test_actor
|
||||||
|
|
||||||
|
""" Tests for scopes.core.actor"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from scopes.core import actor
|
||||||
|
from scopes.util import keywords as k
|
||||||
|
|
||||||
|
collector = []
|
||||||
|
|
||||||
|
def show(msg):
|
||||||
|
collector.append(msg)
|
||||||
|
|
||||||
|
async def handle_basic(msg):
|
||||||
|
match msg:
|
||||||
|
case k.show: show(msg)
|
||||||
|
case k.switch: return handle_alt
|
||||||
|
case _: show(msg)
|
||||||
|
|
||||||
|
async def handle_alt(msg):
|
||||||
|
match msg:
|
||||||
|
case k.show: show('switched')
|
||||||
|
case _: await asyncio.sleep(0.01); show(msg)
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_actor():
|
||||||
|
async with asyncio.TaskGroup() as group:
|
||||||
|
q1 = actor.create(handle_basic, group)
|
||||||
|
q2 = actor.create(handle_basic, group)
|
||||||
|
await q1.put("actor 1")
|
||||||
|
await q1.put(k.show)
|
||||||
|
await q1.put(k.switch)
|
||||||
|
await q1.put(k.show)
|
||||||
|
await q1.put("actor 1: Goodbye")
|
||||||
|
await q2.put("actor 2")
|
||||||
|
await q2.put(k.show)
|
||||||
|
await q1.put(actor.quit)
|
||||||
|
await q2.put(actor.quit)
|
||||||
|
#await asyncio.wait([t1, t2])
|
||||||
|
#print(collector)
|
||||||
|
assert len(collector) == 6
|
||||||
|
assert collector[-1] == 'actor 1: Goodbye'
|
||||||
|
|
||||||
|
|
@ -59,5 +59,5 @@ def test_tracking(storage, caplog):
|
||||||
tr['timeStamp'] = '...'
|
tr['timeStamp'] = '...'
|
||||||
assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'},
|
assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'},
|
||||||
'data': {'activity': 'testing'}, 'timeStamp': '...'}
|
'data': {'activity': 'testing'}, 'timeStamp': '...'}
|
||||||
print('\nLogs: ', caplog.text)
|
#print('\nLogs: ', caplog.text)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,6 @@
|
||||||
# py-scopes: tests.test_util
|
# py-scopes: tests.test_util
|
||||||
|
|
||||||
"""Tests for scopes.util package"""
|
"""Tests for the scopes.util package"""
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue