diff --git a/server/python/rpc-iterator-test.py b/server/python/rpc-iterator-test.py index 6709efd09..cf7e8fe75 100644 --- a/server/python/rpc-iterator-test.py +++ b/server/python/rpc-iterator-test.py @@ -1,26 +1,45 @@ -import sys import asyncio +import rpc from rpc_reader import prepare_peer_readloop +import traceback + +class Bar: + pass async def main(): peer, peerReadLoop = await prepare_peer_readloop(loop, 4, 3) peer.params['foo'] = 3 + jsoncopy = {} + jsoncopy[rpc.RpcPeer.PROPERTY_JSON_COPY_SERIALIZE_CHILDREN] = True + jsoncopy['bar'] = Bar() + peer.params['bar'] = jsoncopy - reader, writer = await asyncio.open_connection( - '127.0.0.1', 6666) + # reader, writer = await asyncio.open_connection( + # '127.0.0.1', 6666) - writer.write(bytes('abcd', 'utf8')) + # writer.write(bytes('abcd', 'utf8')) - async def ticker(delay, to): - for i in range(to): - # print(i) - yield i - await asyncio.sleep(delay) + # async def ticker(delay, to): + # for i in range(to): + # # print(i) + # yield i + # await asyncio.sleep(delay) - peer.params['ticker'] = ticker(0, 3) + # peer.params['ticker'] = ticker(0, 3) print('python starting') - await peerReadLoop() + # await peerReadLoop() + asyncio.ensure_future(peerReadLoop()) + + # print('getting param') + test = await peer.getParam('test') + print(test) + try: + async for c in test: + print(c) + except: + traceback.print_exc() + print('all done iterating') loop = asyncio.new_event_loop() loop.run_until_complete(main()) diff --git a/server/python/rpc.py b/server/python/rpc.py index 83553e085..5ee0cd606 100644 --- a/server/python/rpc.py +++ b/server/python/rpc.py @@ -62,6 +62,21 @@ class RpcProxy(object): self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES] = proxyProps self.__dict__['__proxy_oneway_methods'] = proxyOneWayMethods + def __aiter__(self): + if self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES] and 'Symbol(Symbol.asyncIterator)' in self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES]: + return self + raise Exception('RpcProxy is not an async iterable') + + async def __anext__(self): + if self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES] and 'Symbol(Symbol.asyncIterator)' in self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES]: + try: + return await RpcProxyMethod(self, self.__dict__[RpcPeer.PROPERTY_PROXY_PROPERTIES]['Symbol(Symbol.asyncIterator)']['next'])() + except RPCResultError as e: + if e.name == 'StopAsyncIteration': + raise StopAsyncIteration() + raise + raise Exception('RpcProxy is not an async iterable') + def __getattr__(self, name): if name == '__proxy_finalizer_id': return self.dict['__proxy_entry']['finalizerId'] @@ -88,6 +103,7 @@ class RpcProxy(object): class RpcPeer: RPC_RESULT_ERROR_NAME = 'RPCResultError' PROPERTY_PROXY_PROPERTIES = '__proxy_props' + PROPERTY_JSON_COPY_SERIALIZE_CHILDREN = '__json_copy_serialize_children' def __init__(self, send: Callable[[object, Callable[[Exception], None], Dict], None]) -> None: self.send = send @@ -182,7 +198,7 @@ class RpcPeer: props['Symbol(Symbol.asyncIterator)'] = { 'next': '__anext__', 'throw': 'athrow', - 'return': 'asend', + 'return': 'aclose', } return props @@ -190,6 +206,12 @@ class RpcPeer: return not value or (type(value) in jsonSerializable) def serialize(self, value, serializationContext: Dict): + if type(value) == dict and value.get(RpcPeer.PROPERTY_JSON_COPY_SERIALIZE_CHILDREN, None): + ret = {} + for (key, val) in value.items(): + ret[key] = self.serialize(val, serializationContext) + return ret + if (RpcPeer.isTransportSafe(value)): return value @@ -289,6 +311,13 @@ class RpcPeer: if type(value) != dict: return value + copySerializeChildren = value.get(RpcPeer.PROPERTY_JSON_COPY_SERIALIZE_CHILDREN, None) + if copySerializeChildren: + ret = {} + for (key, val) in value.items(): + ret[key] = self.deserialize(val, deserializationContext) + return ret + __remote_proxy_id = value.get('__remote_proxy_id', None) __remote_proxy_finalizer_id = value.get( '__remote_proxy_finalizer_id', None) @@ -301,7 +330,7 @@ class RpcPeer: '__remote_proxy_oneway_methods', None) if __remote_constructor_name == RpcPeer.RPC_RESULT_ERROR_NAME: - return self.deserializeError(__serialized_value); + return RpcPeer.deserializeError(__serialized_value) if __remote_proxy_id: weakref = self.remoteWeakProxies.get('__remote_proxy_id', None) @@ -366,6 +395,9 @@ class RpcPeer: for arg in (message['args'] or []): args.append(self.deserialize(arg, deserializationContext)) + # if method == 'asend' and hasattr(target, '__aiter__') and hasattr(target, '__anext__') and not len(args): + # args.append(None) + value = None if method: if not hasattr(target, method): diff --git a/server/python/rpc_reader.py b/server/python/rpc_reader.py index af4366836..d0407b159 100644 --- a/server/python/rpc_reader.py +++ b/server/python/rpc_reader.py @@ -41,12 +41,20 @@ async def readLoop(loop, peer: rpc.RpcPeer, reader: asyncio.StreamReader): 'buffers': [] } + if isinstance(reader, asyncio.StreamReader): + async def read(n): + return await reader.readexactly(n) + else: + async def read(n): + return await reader.read(n) + + while True: - lengthBytes = await reader.read(4) - typeBytes = await reader.read(1) + lengthBytes = await read(4) + typeBytes = await read(1) type = typeBytes[0] length = int.from_bytes(lengthBytes, 'big') - data = await reader.read(length - 1) + data = await read(length - 1) if type == 1: deserializationContext['buffers'].append(data) @@ -73,6 +81,7 @@ async def prepare_peer_readloop(loop: AbstractEventLoop, readFd: int = None, wri except Exception as e: if reject: reject(e) + return None else: def write(buffers, reject): try: diff --git a/server/src/rpc.ts b/server/src/rpc.ts index 60014ff86..0dd9a5871 100644 --- a/server/src/rpc.ts +++ b/server/src/rpc.ts @@ -200,11 +200,17 @@ class RpcProxy implements PrimitiveProxyHandler { }); const asyncIterator = this.proxyProps?.[Symbol.asyncIterator.toString()]; - if (!asyncIterator || method !== asyncIterator.next) + if (!asyncIterator || (method !== asyncIterator.next && method !== asyncIterator.return)) return pendingResult; return pendingResult .then(value => { + if (method === asyncIterator.return) { + return { + done: true, + value: undefined, + } + } return ({ value, done: false, @@ -686,11 +692,19 @@ export class RpcPeer { case 'param': { const rpcParam = message as RpcParam; const serializationContext: any = {}; - const result: RpcResult = { - type: 'result', - id: rpcParam.id, - result: this.serialize(this.params[rpcParam.param], serializationContext) - }; + let result: RpcResult; + try { + result = { + type: 'result', + id: rpcParam.id, + result: this.serialize(this.params[rpcParam.param], serializationContext) + }; + } + catch (e) { + // console.error('failure', rpcApply.method, e); + this.createErrorResult(result, e); + } + this.send(result, undefined, serializationContext); break; } diff --git a/server/test/rpc-python-test.ts b/server/test/rpc-python-test.ts index 9962ab8eb..4e1f838da 100644 --- a/server/test/rpc-python-test.ts +++ b/server/test/rpc-python-test.ts @@ -20,14 +20,25 @@ async function main() { const rpcPeer = createDuplexRpcPeer('node', 'python', cp.stdio[3] as Readable, cp.stdio[4] as Writable); - const foo = await rpcPeer.getParam('foo'); - assert.equal(foo, 3); - - const ticker = await rpcPeer.getParam('ticker'); - for await (const v of ticker) { - console.log(v); + async function* test() { + yield 1; + yield 2; + yield 3; } - process.exit(); + + rpcPeer.params['test'] = test(); + + // const foo = await rpcPeer.getParam('foo'); + // assert.equal(foo, 3); + + // const bar = await rpcPeer.getParam('bar'); + // console.log(bar); + + // const ticker = await rpcPeer.getParam('ticker'); + // for await (const v of ticker) { + // console.log(v); + // } + // process.exit(); } main();