From 888b906781232adc781fb5815016a69994148d59 Mon Sep 17 00:00:00 2001 From: Jason Paryani Date: Mon, 2 Dec 2013 17:38:32 -0800 Subject: [PATCH] Remove all public references to EventLoop to match changes to async API --- capnp/__init__.py | 2 +- capnp/capabilityHelper.h | 16 +--- capnp/capnp.pyx | 144 ++++++++++++++++++-------------- capnp/capnp_cpp.pxd | 27 +++--- capnp/rpcHelper.h | 4 +- examples/example_capability.py | 10 +-- examples/example_client.py | 11 +-- examples/example_server.py | 9 +- test/test_capability.py | 62 ++++++-------- test/test_capability_context.py | 62 ++++++-------- test/test_rpc.py | 16 ++-- 11 files changed, 168 insertions(+), 195 deletions(-) diff --git a/capnp/__init__.py b/capnp/__init__.py index 06e15fe..cc982ac 100644 --- a/capnp/__init__.py +++ b/capnp/__init__.py @@ -33,7 +33,7 @@ Example Usage:: """ from .version import version as __version__ from .capnp import * -from .capnp import _DynamicStructReader, _DynamicStructBuilder, _DynamicResizableListBuilder, _DynamicListReader, _DynamicListBuilder, _DynamicOrphan, _DynamicResizableListBuilder, _MallocMessageBuilder, _PackedFdMessageReader, _StreamFdMessageReader, _write_message_to_fd, _write_packed_message_to_fd, _StructModule, _InterfaceModule, _DynamicCapabilityClient, _CapabilityClient +from .capnp import _DynamicStructReader, _DynamicStructBuilder, _DynamicResizableListBuilder, _DynamicListReader, _DynamicListBuilder, _DynamicOrphan, _DynamicResizableListBuilder, _MallocMessageBuilder, _PackedFdMessageReader, _StreamFdMessageReader, _write_message_to_fd, _write_packed_message_to_fd, _StructModule, _InterfaceModule, _DynamicCapabilityClient, _CapabilityClient, _EventLoop del capnp add_import_hook() # enable import hook by default diff --git a/capnp/capabilityHelper.h b/capnp/capabilityHelper.h index 69a29d5..23107ed 100644 --- a/capnp/capabilityHelper.h +++ b/capnp/capabilityHelper.h @@ -76,18 +76,6 @@ void wrapRemoteCall(PyObject * func, capnp::Response & arg check_py_error(); } -::kj::Promise evalLater(kj::EventLoop & loop, PyObject * func) { - return loop.evalLater([func]() { return wrapPyFunc(func, NULL); } ); -} - -::kj::Promise there(kj::EventLoop & loop, kj::Promise & promise, PyObject * func, PyObject * error_func) { - if(error_func == Py_None) - return loop.there(kj::mv(promise), [func](PyObject * arg) { return wrapPyFunc(func, arg); } ); - else - return loop.there(kj::mv(promise), [func](PyObject * arg) { return wrapPyFunc(func, arg); } - , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); -} - ::kj::Promise then(kj::Promise & promise, PyObject * func, PyObject * error_func) { if(error_func == Py_None) return promise.then([func](PyObject * arg) { return wrapPyFunc(func, arg); } ); @@ -134,8 +122,8 @@ public: } }; -capnp::DynamicCapability::Client new_client(capnp::InterfaceSchema & schema, PyObject * server, kj::EventLoop & loop) { - return capnp::DynamicCapability::Client(kj::heap(schema, server), loop); +capnp::DynamicCapability::Client new_client(capnp::InterfaceSchema & schema, PyObject * server) { + return capnp::DynamicCapability::Client(kj::heap(schema, server)); } capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject * server) { return capnp::DynamicValue::Reader(kj::heap(schema, server)); diff --git a/capnp/capnp.pyx b/capnp/capnp.pyx index 0d1b61e..9084bcf 100644 --- a/capnp/capnp.pyx +++ b/capnp/capnp.pyx @@ -1163,6 +1163,61 @@ cdef class _DynamicObjectBuilder: return _DynamicStructBuilder()._init(self.thisptr.getAs(s.thisptr), self._parent) +cdef class _EventLoop: + cdef UnixEventLoop * thisptr + + def __init__(self): + self._init() + + cdef _init(self) except +reraise_kj_exception: + self.thisptr = new UnixEventLoop() + + def __dealloc__(self): + del self.thisptr + + def delete(self): + del self.thisptr + + # cpdef evalLater(self, func): + # Py_INCREF(func) + # return _Promise()._init(capnp.evalLater(self.thisptr, func)) + + # cpdef wait(self, _PromiseTypes promise) except +reraise_kj_exception: + # if promise.is_consumed: + # raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') + + # ret = None + # if _PromiseTypes is _RemotePromise: + # ret = _Response()._init_child(self.thisptr.wait_remote(moveRemotePromise(deref(promise.thisptr))), promise._parent) + # elif _PromiseTypes is _VoidPromise: + # self.thisptr.wait_void(moveVoidPromise(deref(promise.thisptr))) + # elif _PromiseTypes is PromiseFulfillerPair: + # self.thisptr.wait_void(moveVoidPromise(deref(promise.thisptr).promise)) + # elif _PromiseTypes is _Promise: + # ret = self.thisptr.wait(movePromise(deref(promise.thisptr))) + # else: + # raise ValueError("Not a valid promise type") + + # promise.is_consumed = True + + # return ret + + # cpdef there(self, Promise promise, object func, object error_func=None): + # if promise.is_consumed: + # raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') + + # Py_INCREF(func) + # Py_INCREF(error_func) + # return Promise()._init(capnp.there(self.thisptr, deref(promise.thisptr), func, error_func)) + +DEFAULT_EVENT_LOOP = _EventLoop() + +def remove_event_loop(): + global DEFAULT_EVENT_LOOP + + DEFAULT_EVENT_LOOP.delete() + DEFAULT_EVENT_LOOP = None + cdef class _CallContext: cdef CallContext * thisptr @@ -1318,40 +1373,6 @@ cdef class _RemotePromise: def to_dict(self, verbose=False): return _to_dict(self, verbose) -cdef class EventLoop: - cdef UnixEventLoop thisptr - cpdef evalLater(self, func): - Py_INCREF(func) - return _Promise()._init(capnp.evalLater(self.thisptr, func)) - - cpdef wait(self, _PromiseTypes promise) except +reraise_kj_exception: - if promise.is_consumed: - raise ValueError('Promise was already used in a consuming operation. You can no longer use this Promise object') - - ret = None - if _PromiseTypes is _RemotePromise: - ret = _Response()._init_child(self.thisptr.wait_remote(moveRemotePromise(deref(promise.thisptr))), promise._parent) - elif _PromiseTypes is _VoidPromise: - self.thisptr.wait_void(moveVoidPromise(deref(promise.thisptr))) - elif _PromiseTypes is PromiseFulfillerPair: - self.thisptr.wait_void(moveVoidPromise(deref(promise.thisptr).promise)) - elif _PromiseTypes is _Promise: - ret = self.thisptr.wait(movePromise(deref(promise.thisptr))) - else: - raise ValueError("Not a valid promise type") - - promise.is_consumed = True - - return ret - - # cpdef there(self, Promise promise, object func, object error_func=None): - # if promise.is_consumed: - # raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') - - # Py_INCREF(func) - # Py_INCREF(error_func) - # return Promise()._init(capnp.there(self.thisptr, deref(promise.thisptr), func, error_func)) - cdef class _Request(_DynamicStructBuilder): cdef Request * thisptr_child @@ -1392,23 +1413,21 @@ cdef class _DynamicCapabilityServer: cdef class _DynamicCapabilityClient: cdef C_DynamicCapability.Client thisptr - cdef public object _event_loop, _server, _parent + cdef public object _server, _parent cdef _init(self, C_DynamicCapability.Client other, object parent): self.thisptr = other self._parent = parent return self - cdef _init_vals(self, schema, server, event_loop): + cdef _init_vals(self, schema, server): cdef _InterfaceSchema s if hasattr(schema, 'schema'): s = schema.schema else: s = schema - cdef EventLoop loop = event_loop - self._event_loop = event_loop - self.thisptr = new_client(s.thisptr, server, loop.thisptr) + self.thisptr = new_client(s.thisptr, server) self._server = server return self @@ -1523,24 +1542,25 @@ cdef class Restorer: cdef class _TwoPartyVatNetwork: cdef Own[C_TwoPartyVatNetwork] thisptr - cdef _init(self, EventLoop loop, AsyncIoStream & stream, Side side): - self.thisptr = makeTwoPartyVatNetwork(loop.thisptr, stream, side) + cdef _init(self, AsyncIoStream & stream, Side side): + self.thisptr = makeTwoPartyVatNetwork(stream, side) return self cdef class RpcClient: cdef RpcSystem * thisptr cdef public _TwoPartyVatNetwork network - cdef public object loop, restorer, stream + cdef public object restorer, _stream + cdef public _FdAsyncIoStream stream - def __init__(self, EventLoop loop, FdAsyncIoStream stream, Restorer restorer=None): - self.loop = loop - self.stream = stream - self.network = _TwoPartyVatNetwork()._init(loop, deref(stream.thisptr), capnp.CLIENT) + def __init__(self, stream, Restorer restorer=None): + self._stream = stream + self.stream = _FdAsyncIoStream(stream.fileno()) + self.network = _TwoPartyVatNetwork()._init(deref(self.stream.thisptr), capnp.CLIENT) if restorer is None: - self.thisptr = new RpcSystem(makeRpcClient(deref(self.network.thisptr), loop.thisptr)) + self.thisptr = new RpcSystem(makeRpcClient(deref(self.network.thisptr))) else: self.restorer = restorer - self.thisptr = new RpcSystem(makeRpcClientWithRestorer(deref(self.network.thisptr), loop.thisptr, deref(restorer.thisptr))) + self.thisptr = new RpcSystem(makeRpcClientWithRestorer(deref(self.network.thisptr), deref(restorer.thisptr))) def __dealloc__(self): del self.thisptr @@ -1569,24 +1589,25 @@ cdef class RpcClient: cdef class RpcServer: cdef RpcSystem * thisptr cdef public _TwoPartyVatNetwork network - cdef public object loop, restorer, stream + cdef public object restorer, _stream + cdef public _FdAsyncIoStream stream - def __init__(self, EventLoop loop, FdAsyncIoStream stream, Restorer restorer): - self.loop = loop - self.stream = stream + def __init__(self, stream, Restorer restorer): + self._stream = stream + self.stream = _FdAsyncIoStream(stream.fileno()) self.restorer = restorer - self.network = _TwoPartyVatNetwork()._init(loop, deref(stream.thisptr), capnp.SERVER) - self.thisptr = new RpcSystem(makeRpcServer(deref(self.network.thisptr), deref(restorer.thisptr), loop.thisptr)) + self.network = _TwoPartyVatNetwork()._init(deref(self.stream.thisptr), capnp.SERVER) + self.thisptr = new RpcSystem(makeRpcServer(deref(self.network.thisptr), deref(restorer.thisptr))) def __dealloc__(self): del self.thisptr def run_forever(self): - self.loop.wait(_VoidPromise()._init(deref(self.network.thisptr).onDisconnect())) + return _VoidPromise()._init(deref(self.network.thisptr).onDisconnect()) # TODO: add restore functionality here? -cdef class FdAsyncIoStream: +cdef class _FdAsyncIoStream: cdef Own[AsyncIoStream] thisptr def __init__(self, int fd): @@ -1596,11 +1617,8 @@ cdef class PromiseFulfillerPair: cdef Own[C_PromiseFulfillerPair] thisptr cdef public bint is_consumed - def __init__(self, EventLoop loop=None): - if loop is None: - self.thisptr = copyPromiseFulfillerPair(newPromiseAndFulfiller()) - else: - self.thisptr = copyPromiseFulfillerPair(newPromiseAndFulfiller(loop.thisptr)) + def __init__(self): + self.thisptr = copyPromiseFulfillerPair(newPromiseAndFulfiller()) self.is_consumed = False cpdef fulfill(self): @@ -1870,8 +1888,8 @@ class _InterfaceModule(object): def __init__(self, schema): self.schema = schema - def _new_client(self, server, loop): - return _DynamicCapabilityClient()._init_vals(self.schema, server, loop) + def _new_client(self, server): + return _DynamicCapabilityClient()._init_vals(self.schema, server) def new_server(self, server): return _DynamicCapabilityServer(self.schema, server) diff --git a/capnp/capnp_cpp.pxd b/capnp/capnp_cpp.pxd index f780e95..654c22b 100644 --- a/capnp/capnp_cpp.pxd +++ b/capnp/capnp_cpp.pxd @@ -35,7 +35,7 @@ cdef extern from "kj/exception.h" namespace " ::kj": cdef extern from "kj/memory.h" namespace " ::kj": cdef cppclass Own[T]: T& operator*() - Own[TwoPartyVatNetwork] makeTwoPartyVatNetwork" ::kj::heap< ::capnp::TwoPartyVatNetwork>"(EventLoop &, AsyncIoStream& stream, Side) + Own[TwoPartyVatNetwork] makeTwoPartyVatNetwork" ::kj::heap< ::capnp::TwoPartyVatNetwork>"(AsyncIoStream& stream, Side) Own[PromiseFulfillerPair] copyPromiseFulfillerPair" ::kj::heap< ::kj::PromiseFulfillerPair >"(PromiseFulfillerPair&) cdef extern from "kj/string-tree.h" namespace " ::kj": @@ -206,8 +206,8 @@ cdef extern from "capnp/rpc-twoparty.h" namespace " ::capnp": TwoPartyVatNetwork(EventLoop &, AsyncIoStream& stream, Side) VoidPromise onDisconnect() VoidPromise onDrain() - RpcSystem makeRpcServer(TwoPartyVatNetwork&, PyRestorer&, EventLoop&) - RpcSystem makeRpcClient(TwoPartyVatNetwork&, EventLoop&) + RpcSystem makeRpcServer(TwoPartyVatNetwork&, PyRestorer&) + RpcSystem makeRpcClient(TwoPartyVatNetwork&) cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass Request" ::capnp::Request< ::capnp::DynamicStruct, ::capnp::DynamicStruct>": @@ -237,13 +237,13 @@ cdef extern from "fixMaybe.h": char * getEnumString(Request val) cdef extern from "capabilityHelper.h": - PyPromise evalLater(EventLoop &, PyObject * func) - PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) + # PyPromise evalLater(EventLoop &, PyObject * func) + # PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) PyPromise then(PyPromise & promise, PyObject * func, PyObject * error_func) VoidPromise then(RemotePromise & promise, PyObject * func, PyObject * error_func) cppclass PythonInterfaceDynamicImpl: PythonInterfaceDynamicImpl(PyObject *) - DynamicCapability.Client new_client(InterfaceSchema&, PyObject *, EventLoop&) + DynamicCapability.Client new_client(InterfaceSchema&, PyObject *) DynamicValueForward.Reader new_server(InterfaceSchema&, PyObject *) Capability.Client server_to_client(InterfaceSchema&, PyObject *) PyPromise convert_to_pypromise(RemotePromise&) @@ -253,7 +253,7 @@ cdef extern from "rpcHelper.h": PyRestorer(PyObject *, StructSchema&) Capability.Client restoreHelper(RpcSystem&, MessageBuilder&) Capability.Client restoreHelper(RpcSystem&, MessageReader&) - RpcSystem makeRpcClientWithRestorer(TwoPartyVatNetwork&, EventLoop&, PyRestorer&) + RpcSystem makeRpcClientWithRestorer(TwoPartyVatNetwork&, PyRestorer&) cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass DynamicEnum: @@ -368,12 +368,12 @@ cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass EventLoop: EventLoop() # Promise[void] yield_end'yield'() - object wait(PyPromise) except +reraise_kj_exception - Response wait_remote'wait'(RemotePromise) - void wait_void'wait'(VoidPromise) - object there(PyPromise) except +reraise_kj_exception - PyPromise evalLater(PyObject * func) - PyPromise there(PyPromise, PyObject * func) + # object wait(PyPromise) except +reraise_kj_exception + # Response wait_remote'wait'(RemotePromise) + # void wait_void'wait'(VoidPromise) + # object there(PyPromise) except +reraise_kj_exception + # PyPromise evalLater(PyObject * func) + # PyPromise there(PyPromise, PyObject * func) cdef cppclass SimpleEventLoop(EventLoop): pass cdef cppclass PromiseFulfiller: @@ -382,7 +382,6 @@ cdef extern from "kj/async.h" namespace " ::kj": VoidPromise promise Own[PromiseFulfiller] fulfiller PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller"() - PromiseFulfillerPair newPromiseAndFulfiller" ::kj::newPromiseAndFulfiller"(EventLoop&) cdef extern from "kj/async-unix.h" namespace " ::kj": cdef cppclass UnixEventLoop(EventLoop): diff --git a/capnp/rpcHelper.h b/capnp/rpcHelper.h index 8ec6daf..d538799 100644 --- a/capnp/rpcHelper.h +++ b/capnp/rpcHelper.h @@ -52,8 +52,8 @@ template capnp::RpcSystem makeRpcClientWithRestorer( capnp::VatNetwork& network, - const kj::EventLoop& eventLoop, PyRestorer& restorer) { + PyRestorer& restorer) { using namespace capnp; return RpcSystem(network, - kj::Maybe&>(restorer), eventLoop); + kj::Maybe&>(restorer)); } diff --git a/examples/example_capability.py b/examples/example_capability.py index db27988..9273024 100644 --- a/examples/example_capability.py +++ b/examples/example_capability.py @@ -15,22 +15,20 @@ def example_simple_rpc(): def _restore(ref_id): return capability.TestInterface.new_server(Server(100)) - loop = capnp.EventLoop() + # loop = capnp.EventLoop() read, write = socket.socketpair(socket.AF_UNIX) - read_stream = capnp.FdAsyncIoStream(read.fileno()) - write_stream = capnp.FdAsyncIoStream(write.fileno()) restorer = capnp.Restorer(capability.TestSturdyRefObjectId, _restore) - server = capnp.RpcServer(loop, write_stream, restorer) - client = capnp.RpcClient(loop, read_stream) + server = capnp.RpcServer(write, restorer) + client = capnp.RpcClient(read) ref = capability.TestSturdyRefObjectId.new_message() cap = client.restore(ref.as_reader()) cap = cap.cast_as(capability.TestInterface) remote = cap.foo(i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '125' diff --git a/examples/example_client.py b/examples/example_client.py index 740b125..d5ce706 100644 --- a/examples/example_client.py +++ b/examples/example_client.py @@ -5,20 +5,15 @@ import test_capnp import socket def example_client(): - loop = capnp.EventLoop() - c = socket.create_connection(('localhost', 49999)) - read_stream = capnp.FdAsyncIoStream(c.fileno()) - client = capnp.RpcClient(loop, read_stream) + client = capnp.RpcClient(c) - ref = test_capnp.TestSturdyRefObjectId.new_message() - ref.tag = 'testInterface' - cap = client.restore(ref) + cap = client.restore(test_capnp.TestSturdyRefObjectId.new_message(tag='testInterface')) cap = cap.cast_as(test_capnp.TestInterface) remote = cap.foo(i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '125' c.close() diff --git a/examples/example_server.py b/examples/example_server.py index 932bc9f..4a12060 100644 --- a/examples/example_server.py +++ b/examples/example_server.py @@ -12,7 +12,10 @@ class Server: return str(i * 5 + self.val) def restore(ref_id): - return test_capnp.TestInterface.new_server(Server(100)) + if ref_id.tag == 'testInterface': + return test_capnp.TestInterface.new_server(Server(100)) + else: + raise ValueError('invalid ref') def example_server(host='localhost', port=49999): backlog = 1 @@ -21,13 +24,11 @@ def example_server(host='localhost', port=49999): s.bind((host,port)) s.listen(backlog) - loop = capnp.EventLoop() while 1: try: (clientsocket, address) = s.accept() - stream = capnp.FdAsyncIoStream(clientsocket.fileno()) restorer = capnp.Restorer(test_capnp.TestSturdyRefObjectId, restore) - server = capnp.RpcServer(loop, stream, restorer) + server = capnp.RpcServer(clientsocket, restorer) server.run_forever() except KeyboardInterrupt: diff --git a/test/test_capability.py b/test/test_capability.py index 1028687..75a468b 100644 --- a/test/test_capability.py +++ b/test/test_capability.py @@ -30,15 +30,13 @@ class PipelineServer: return inCap.foo(i=n).then(_then) def test_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestInterface._new_client(Server()) req = client._request('foo') req.i = 5 remote = req.send() - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' @@ -46,7 +44,7 @@ def test_client(capability): req.i = 5 remote = req.send() - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' @@ -64,43 +62,41 @@ def test_client(capability): req.baz = 1 def test_simple_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestInterface._new_client(Server()) remote = client._send('foo', i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(i=5, j=True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.foo(5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(5, True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.foo(5, j=True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.buz(capability.TestSturdyRefHostId.new_message(host='localhost')) - response = loop.wait(remote) + response = remote.wait() assert response.x == 'localhost_test' @@ -120,20 +116,18 @@ def test_simple_client(capability): remote = client.foo(baz=5) def test_pipeline(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(PipelineServer(), loop) - foo_client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestPipeline._new_client(PipelineServer()) + foo_client = capability.TestInterface._new_client(Server()) remote = client.getCap(n=5, inCap=foo_client) outCap = remote.outBox.cap pipelinePromise = outCap.foo(i=10) - response = loop.wait(pipelinePromise) + response = pipelinePromise.wait() assert response.x == '150' - response = loop.wait(remote) + response = remote.wait() assert response.s == '26_foo' class BadServer: @@ -147,13 +141,11 @@ class BadServer: return str(i * 5 + extra + self.val), 10 # returning too many args def test_exception_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestInterface._new_client(BadServer()) remote = client._send('foo', i=5) with pytest.raises(capnp.KjException): - loop.wait(remote) + remote.wait() class BadPipelineServer: def getCap(self, n, inCap, _results, **kwargs): @@ -166,23 +158,19 @@ class BadPipelineServer: return inCap.foo(i=n).then(_then, _error) def test_exception_chain(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(BadPipelineServer(), loop) - foo_client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestPipeline._new_client(BadPipelineServer()) + foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) try: - loop.wait(remote) + remote.wait() except Exception as e: assert 'test was a success' in str(e) def test_pipeline_exception(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(BadPipelineServer(), loop) - foo_client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestPipeline._new_client(BadPipelineServer()) + foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) @@ -193,12 +181,10 @@ def test_pipeline_exception(capability): loop.wait(pipelinePromise) with pytest.raises(Exception): - loop.wait(remote) + remote.wait() def test_casting(capability): - loop = capnp.EventLoop() - - client = capability.TestExtends._new_client(Server(), loop) + client = capability.TestExtends._new_client(Server()) client2 = client.upcast(capability.TestInterface) client3 = client2.cast_as(capability.TestInterface) diff --git a/test/test_capability_context.py b/test/test_capability_context.py index 7c8a3f9..18cccf2 100644 --- a/test/test_capability_context.py +++ b/test/test_capability_context.py @@ -30,15 +30,13 @@ class PipelineServer: return context.params.inCap.foo(i=context.params.n).then(_then) def test_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestInterface._new_client(Server()) req = client._request('foo') req.i = 5 remote = req.send() - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' @@ -46,7 +44,7 @@ def test_client(capability): req.i = 5 remote = req.send() - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' @@ -64,43 +62,41 @@ def test_client(capability): req.baz = 1 def test_simple_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestInterface._new_client(Server()) remote = client._send('foo', i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(i=5, j=True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.foo(5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '26' remote = client.foo(5, True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.foo(5, j=True) - response = loop.wait(remote) + response = remote.wait() assert response.x == '27' remote = client.buz(capability.TestSturdyRefHostId.new_message(host='localhost')) - response = loop.wait(remote) + response = remote.wait() assert response.x == 'localhost_test' @@ -120,20 +116,18 @@ def test_simple_client(capability): remote = client.foo(baz=5) def test_pipeline(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(PipelineServer(), loop) - foo_client = capability.TestInterface._new_client(Server(), loop) + client = capability.TestPipeline._new_client(PipelineServer()) + foo_client = capability.TestInterface._new_client(Server()) remote = client.getCap(n=5, inCap=foo_client) outCap = remote.outBox.cap pipelinePromise = outCap.foo(i=10) - response = loop.wait(pipelinePromise) + response = pipelinePromise.wait() assert response.x == '150' - response = loop.wait(remote) + response = remote.wait() assert response.s == '26_foo' class BadServer: @@ -145,13 +139,11 @@ class BadServer: context.results.x2 = 5 # raises exception def test_exception_client(capability): - loop = capnp.EventLoop() - - client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestInterface._new_client(BadServer()) remote = client._send('foo', i=5) with pytest.raises(capnp.KjException): - loop.wait(remote) + remote.wait() class BadPipelineServer: def getCap_context(self, context): @@ -164,23 +156,19 @@ class BadPipelineServer: return context.params.inCap.foo(i=context.params.n).then(_then, _error) def test_exception_chain(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(BadPipelineServer(), loop) - foo_client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestPipeline._new_client(BadPipelineServer()) + foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) try: - loop.wait(remote) + remote.wait() except Exception as e: assert 'test was a success' in str(e) def test_pipeline_exception(capability): - loop = capnp.EventLoop() - - client = capability.TestPipeline._new_client(BadPipelineServer(), loop) - foo_client = capability.TestInterface._new_client(BadServer(), loop) + client = capability.TestPipeline._new_client(BadPipelineServer()) + foo_client = capability.TestInterface._new_client(BadServer()) remote = client.getCap(n=5, inCap=foo_client) @@ -191,12 +179,10 @@ def test_pipeline_exception(capability): loop.wait(pipelinePromise) with pytest.raises(Exception): - loop.wait(remote) + remote.wait() def test_casting(capability): - loop = capnp.EventLoop() - - client = capability.TestExtends._new_client(Server(), loop) + client = capability.TestExtends._new_client(Server()) client2 = client.upcast(capability.TestInterface) client3 = client2.cast_as(capability.TestInterface) diff --git a/test/test_rpc.py b/test/test_rpc.py index b520e55..abfb3cc 100644 --- a/test/test_rpc.py +++ b/test/test_rpc.py @@ -20,21 +20,23 @@ def test_simple_rpc(capability): def _restore(ref_id): return capability.TestInterface.new_server(Server(100)) - loop = capnp.EventLoop() - read, write = socket.socketpair(socket.AF_UNIX) - read_stream = capnp.FdAsyncIoStream(read.fileno()) - write_stream = capnp.FdAsyncIoStream(write.fileno()) restorer = capnp.Restorer(capability.TestSturdyRefObjectId, _restore) - server = capnp.RpcServer(loop, write_stream, restorer) - client = capnp.RpcClient(loop, read_stream) + server = capnp.RpcServer(write, restorer) + client = capnp.RpcClient(read) ref = capability.TestSturdyRefObjectId.new_message() cap = client.restore(ref) cap = cap.cast_as(capability.TestInterface) remote = cap.foo(i=5) - response = loop.wait(remote) + response = remote.wait() assert response.x == '125' + +# def test_custom_event_loop(capability): +# capnp.remove_event_loop() +# capnp.DEFAULT_EVENT_LOOP = capnp._EventLoop() + +# test_simple_rpc(capability) \ No newline at end of file