diff --git a/capnp/async_cpp.pxd b/capnp/async_cpp.pxd index 3817010..21d3a8b 100644 --- a/capnp/async_cpp.pxd +++ b/capnp/async_cpp.pxd @@ -16,14 +16,3 @@ cdef extern from "kj/async.h" namespace " ::kj": ctypedef Promise[PyObject *] PyPromise ctypedef Promise[void] VoidPromise - -cdef extern from "kj/async.h" namespace " ::kj": - cdef cppclass EventLoop: - EventLoop() - # Promise[void] yield_end'yield'() - object wait(PyPromise) except+ - object there(PyPromise) except+ - PyPromise evalLater(PyObject * func) - PyPromise there(PyPromise, PyObject * func) - cdef cppclass SimpleEventLoop(EventLoop): - pass \ No newline at end of file diff --git a/capnp/capabilityHelper.h b/capnp/capabilityHelper.h index 0c9023a..993e031 100644 --- a/capnp/capabilityHelper.h +++ b/capnp/capabilityHelper.h @@ -1,3 +1,5 @@ +#pragma once + #include "capnp/dynamic.h" #include #include "Python.h" @@ -97,6 +99,10 @@ capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject return capnp::DynamicValue::Reader(kj::heap(schema, server)); } +capnp::Capability::Client server_to_client(capnp::InterfaceSchema & schema, PyObject * server) { + return kj::heap(schema, server); +} + ::kj::Promise convert_to_pypromise(capnp::RemotePromise & promise) { return promise.then([](capnp::Response&& response) { return wrap_dynamic_struct_reader(response); } ); -} \ No newline at end of file +} diff --git a/capnp/capnp.pyx b/capnp/capnp.pyx index ccc3fba..f1aea89 100644 --- a/capnp/capnp.pyx +++ b/capnp/capnp.pyx @@ -1,7 +1,7 @@ # capnp.pyx # distutils: language = c++ # distutils: extra_compile_args = --std=c++11 -# distutils: libraries = capnpc capnp +# distutils: libraries = capnpc capnp capnp-rpc # cython: c_string_type = str # cython: c_string_encoding = default # cython: embedsignature = True @@ -9,7 +9,7 @@ cimport cython cimport capnp_cpp as capnp cimport schema_cpp -from capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, fixMaybe, getEnumString, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, ObjectPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, new_client, new_server, Request, Response, RemotePromise, convert_to_pypromise, SimpleEventLoop, PyPromise, VoidPromise, CallContext +from capnp_cpp cimport Schema as C_Schema, StructSchema as C_StructSchema, InterfaceSchema as C_InterfaceSchema, DynamicStruct as C_DynamicStruct, DynamicValue as C_DynamicValue, Type as C_Type, DynamicList as C_DynamicList, fixMaybe, getEnumString, SchemaParser as C_SchemaParser, ParsedSchema as C_ParsedSchema, VOID, ArrayPtr, StringPtr, String, StringTree, DynamicOrphan as C_DynamicOrphan, ObjectPointer as C_DynamicObject, DynamicCapability as C_DynamicCapability, new_client, new_server, server_to_client, Request, Response, RemotePromise, convert_to_pypromise, UnixEventLoop, PyPromise, VoidPromise, CallContext, PyRestorer, RpcSystem, makeRpcServer, makeRpcClient, TwoWayPipe as C_TwoWayPipe, newTwoWayPipe, restoreHelper, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side from schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode from cython.operator cimport dereference as deref @@ -73,6 +73,15 @@ cdef public VoidPromise * call_server_method(PyObject * _server, char * _method_ return NULL +cdef public C_Capability.Client * call_py_restorer(PyObject * _restorer, C_DynamicStruct.Reader & _reader) except *: + restorer = _restorer + reader = _DynamicStructReader()._init(_reader, None) + + ret = restorer.restore(reader) + cdef _DynamicCapabilityServer server = ret + + return new C_Capability.Client(server_to_client(server.schema.thisptr, server.server)) + cdef public object wrap_kj_exception(capnp.Exception & exception): return None # TODO @@ -128,6 +137,7 @@ cdef extern from "" namespace "std": VoidPromise moveVoidPromise"std::move"(VoidPromise) RemotePromise moveRemotePromise"std::move"(RemotePromise) CallContext moveCallContext"std::move"(CallContext) + capnp.Own[capnp.AsyncIoStream] moveOwnAsyncIOStream"std::move"(capnp.Own[capnp.AsyncIoStream]) cdef extern from "" namespace " ::capnp": StringTree printStructReader" ::capnp::prettyPrint"(C_DynamicStruct.Reader) @@ -1148,7 +1158,7 @@ cdef class _RemotePromise: return _to_dict(self, verbose) cdef class EventLoop: - cdef SimpleEventLoop thisptr + cdef UnixEventLoop thisptr cpdef evalLater(self, func): Py_INCREF(func) return Promise()._init(capnp.evalLater(self.thisptr, func)) @@ -1253,6 +1263,93 @@ cdef class _DynamicCapabilityClient: return _partial(self._request, short_name) return _partial(self._send, name) +cdef class _CapabilityClient: + cdef C_Capability.Client * thisptr + cdef public object _parent + + cdef _init(self, C_Capability.Client other, object parent): + self.thisptr = new C_Capability.Client(other) + self._parent = parent + return self + + def __dealloc__(self): + del self.thisptr + + cpdef cast_as(self, schema): + cdef _InterfaceSchema s + if hasattr(schema, 'schema'): + s = schema.schema + else: + s = schema + return _DynamicCapabilityClient()._init(self.thisptr.castAs(s.thisptr), self._parent) + +cdef class Restorer: + cdef PyRestorer * thisptr + cdef C_StructSchema schema + + cdef public object restore + + def __init__(self, schema, restore_func): + cdef _StructSchema s + if hasattr(schema, 'schema'): + s = schema.schema + else: + s = schema + + self.schema = s.thisptr + self.restore = restore_func + self.thisptr = new PyRestorer(self, self.schema) + + def __dealloc__(self): + del self.thisptr + +cdef class _TwoPartyVatNetwork: + cdef C_TwoPartyVatNetwork * thisptr + + cdef _init(self, EventLoop loop, capnp.AsyncIoStream & stream, Side side): + self.thisptr = new C_TwoPartyVatNetwork(loop.thisptr, stream, side) + return self + + def __dealloc__(self): + del self.thisptr + +cdef class RpcClient: + cdef RpcSystem * thisptr + cdef public _TwoPartyVatNetwork network + cdef public object loop + + def __init__(self, EventLoop loop, TwoWayPipe pipe): + self.loop = loop + self.network = _TwoPartyVatNetwork()._init(loop, deref(moveOwnAsyncIOStream(pipe.thisptr.ends[0])), capnp.CLIENT) + self.thisptr = new RpcSystem(makeRpcClient(deref(self.network.thisptr), loop.thisptr)) + + def __dealloc__(self): + del self.thisptr + + cpdef restore(self, _DynamicStructReader objectId) except+: + cdef _MessageBuilder builder = objectId._parent + return _CapabilityClient()._init(restoreHelper(deref(self.thisptr), deref(builder.thisptr)), self) + +cdef class RpcServer: + cdef RpcSystem * thisptr + cdef public _TwoPartyVatNetwork network + cdef public object loop, restorer + + def __init__(self, EventLoop loop, Restorer restorer, TwoWayPipe pipe): + self.loop = loop + self.restorer = restorer + self.network = _TwoPartyVatNetwork()._init(loop, deref(moveOwnAsyncIOStream(pipe.thisptr.ends[1])), capnp.SERVER) + self.thisptr = new RpcSystem(makeRpcServer(deref(self.network.thisptr), deref(restorer.thisptr), loop.thisptr)) + + def __dealloc__(self): + del self.thisptr + +cdef class TwoWayPipe: + cdef C_TwoWayPipe thisptr + + def __init__(self): + self.thisptr = newTwoWayPipe() + cdef class _Schema: cdef C_Schema thisptr cdef _init(self, C_Schema other): diff --git a/capnp/capnp_cpp.pxd b/capnp/capnp_cpp.pxd index cc02e75..e37ea5b 100644 --- a/capnp/capnp_cpp.pxd +++ b/capnp/capnp_cpp.pxd @@ -1,7 +1,7 @@ # schema.capnp.cpp.pyx # distutils: language = c++ # distutils: extra_compile_args = --std=c++11 -from schema_cpp cimport Node, Data, StructNode, EnumNode +from schema_cpp cimport Node, Data, StructNode, EnumNode, MessageBuilder from async_cpp cimport PyPromise, VoidPromise, Promise from cpython.ref cimport PyObject @@ -41,6 +41,16 @@ cdef extern from "kj/array.h" namespace " ::kj": T* begin() size_t size() +cdef extern from "kj/async-io.h" namespace " ::kj": + cdef cppclass Own[T]: + T& operator*() + + cdef cppclass AsyncIoStream: + pass + cdef cppclass TwoWayPipe: + Own[AsyncIoStream] * ends + TwoWayPipe newTwoWayPipe() + cdef extern from "capnp/schema.h" namespace " ::capnp": cdef cppclass Schema: Node.Reader getProto() except + @@ -141,11 +151,35 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": DynamicValueForward.Pipeline get(char *) StructSchema getSchema() +cdef extern from "capnp/dynamic.h" namespace " ::capnp": + cdef cppclass DynamicCapability: + cppclass Client: + Client() + Client(Client&) + Client upcast(InterfaceSchema requestedSchema) + InterfaceSchema getSchema() + Request newRequest(char * methodName, uint firstSegmentWordSize) + cdef extern from "capnp/capability.h" namespace " ::capnp": cdef cppclass Response" ::capnp::Response< ::capnp::DynamicStruct>"(DynamicStruct.Reader): Response(Response) cdef cppclass RemotePromise" ::capnp::RemotePromise< ::capnp::DynamicStruct>"(Promise[Response], DynamicStruct.Pipeline): RemotePromise(RemotePromise) + cdef cppclass Capability: + cppclass Client: + Client(Client&) + DynamicCapability.Client castAs"castAs< ::capnp::DynamicCapability>"(InterfaceSchema) + +cdef extern from "capnp/rpc-twoparty.h" namespace " ::capnp": + cdef cppclass RpcSystem" ::capnp::RpcSystem": + RpcSystem(RpcSystem&&) + enum Side" ::capnp::rpc::twoparty::Side": + CLIENT" ::capnp::rpc::twoparty::Side::CLIENT" + SERVER" ::capnp::rpc::twoparty::Side::SERVER" + cdef cppclass TwoPartyVatNetwork: + TwoPartyVatNetwork(EventLoop &, AsyncIoStream& stream, Side) + RpcSystem makeRpcServer(TwoPartyVatNetwork&, PyRestorer&, EventLoop&) + RpcSystem makeRpcClient(TwoPartyVatNetwork&, EventLoop&) cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass Request" ::capnp::Request< ::capnp::DynamicStruct, ::capnp::DynamicStruct>": @@ -160,12 +194,6 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": Maybe[StructSchema.Field] which() RemotePromise send() - cdef cppclass DynamicCapability: - cppclass Client: - Client upcast(InterfaceSchema requestedSchema) - InterfaceSchema getSchema() - Request newRequest(char * methodName, uint firstSegmentWordSize) - cdef extern from "capnp/object.h" namespace " ::capnp": cdef cppclass ObjectPointer: cppclass Reader: @@ -189,8 +217,14 @@ cdef extern from "capabilityHelper.h": PythonInterfaceDynamicImpl(PyObject *) DynamicCapability.Client new_client(InterfaceSchema&, PyObject *, EventLoop&) DynamicValueForward.Reader new_server(InterfaceSchema&, PyObject *) + Capability.Client server_to_client(InterfaceSchema&, PyObject *) PyPromise convert_to_pypromise(RemotePromise&) +cdef extern from "rpcHelper.h": + cdef cppclass PyRestorer: + PyRestorer(PyObject *, StructSchema&) + Capability.Client restoreHelper(RpcSystem&, MessageBuilder&) + cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass DynamicEnum: uint16_t getRaw() @@ -304,3 +338,8 @@ cdef extern from "kj/async.h" namespace " ::kj": PyPromise there(PyPromise, PyObject * func) cdef cppclass SimpleEventLoop(EventLoop): pass + +cdef extern from "kj/async-unix.h" namespace " ::kj": + cdef cppclass UnixEventLoop(EventLoop): + pass + diff --git a/examples/example_capability.capnp b/examples/example_capability.capnp index b947bda..bb93c1d 100644 --- a/examples/example_capability.capnp +++ b/examples/example_capability.capnp @@ -43,3 +43,16 @@ interface TestPipeline { cap @0 :TestInterface; } } + +struct TestSturdyRefHostId { + host @0 :Text; +} + +struct TestSturdyRefObjectId { + tag @0 :Tag; + enum Tag { + testInterface @0; + testExtends @1; + testPipeline @2; + } +} \ No newline at end of file diff --git a/examples/example_capability.py b/examples/example_capability.py index 23059bd..6812846 100644 --- a/examples/example_capability.py +++ b/examples/example_capability.py @@ -4,21 +4,30 @@ import capnp import example_capability_capnp class Server: + def __init__(self, val=1): + self.val = val + def foo(self, context): - context.results.x = str(context.params.i * 5 + 1) + context.results.x = str(context.params.i * 5 + self.val) + +def test_simple_rpc(): + def _restore(ref_id): + return example_capability_capnp.TestInterface.new_server(Server(100)) -def example_client(): loop = capnp.EventLoop() - - client = example_capability_capnp.TestInterface.new_client(Server(), loop) - req = client._request('foo') - req.i = 5 + pipe = capnp.TwoWayPipe() + restorer = capnp.Restorer(example_capability_capnp.TestSturdyRefObjectId, _restore) + server = capnp.RpcServer(loop, restorer, pipe) + client = capnp.RpcClient(loop, pipe) - remote = req.send() + ref = example_capability_capnp.TestSturdyRefObjectId.new_message() + cap = client.restore(ref.as_reader()) + cap = cap.cast_as(example_capability_capnp.TestInterface) + + remote = cap.foo(i=5) response = loop.wait_remote(remote) - print(response.x) + assert response.x == '125' -if __name__ == '__main__': - example_client() +test_simple_rpc() \ No newline at end of file diff --git a/test/test_capability.capnp b/test/test_capability.capnp index 7d59d8f..8ce0030 100644 --- a/test/test_capability.capnp +++ b/test/test_capability.capnp @@ -43,3 +43,16 @@ interface TestPipeline { cap @0 :TestInterface; } } + +struct TestSturdyRefHostId { + host @0 :Text; +} + +struct TestSturdyRefObjectId { + tag @0 :Tag; + enum Tag { + testInterface @0; + testExtends @1; + testPipeline @2; + } +} \ No newline at end of file