diff --git a/capnp/capnp.pyx b/capnp/capnp.pyx index f1aea89..8391a3f 100644 --- a/capnp/capnp.pyx +++ b/capnp/capnp.pyx @@ -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, 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 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, restoreHelper, Capability as C_Capability, TwoPartyVatNetwork as C_TwoPartyVatNetwork, Side, AsyncIoStream_wrapFd, AsyncIoStream, Own from schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode from cython.operator cimport dereference as deref @@ -137,7 +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]) + Own[AsyncIoStream] moveOwnAsyncIOStream"std::move"(Own[AsyncIoStream]) cdef extern from "" namespace " ::capnp": StringTree printStructReader" ::capnp::prettyPrint"(C_DynamicStruct.Reader) @@ -1306,7 +1306,7 @@ cdef class Restorer: cdef class _TwoPartyVatNetwork: cdef C_TwoPartyVatNetwork * thisptr - cdef _init(self, EventLoop loop, capnp.AsyncIoStream & stream, Side side): + cdef _init(self, EventLoop loop, AsyncIoStream & stream, Side side): self.thisptr = new C_TwoPartyVatNetwork(loop.thisptr, stream, side) return self @@ -1318,37 +1318,43 @@ cdef class RpcClient: cdef public _TwoPartyVatNetwork network cdef public object loop - def __init__(self, EventLoop loop, TwoWayPipe pipe): + def __init__(self, EventLoop loop, FdAsyncIoStream stream): self.loop = loop - self.network = _TwoPartyVatNetwork()._init(loop, deref(moveOwnAsyncIOStream(pipe.thisptr.ends[0])), capnp.CLIENT) + self.network = _TwoPartyVatNetwork()._init(loop, deref(stream.thisptr), 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) + cpdef restore(self, objectId) except+: + cdef _MessageBuilder builder + cdef _MessageReader reader + try: + builder = objectId._parent + return _CapabilityClient()._init(restoreHelper(deref(self.thisptr), deref(builder.thisptr)), self) + except: + reader = objectId._parent + return _CapabilityClient()._init(restoreHelper(deref(self.thisptr), deref(reader.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): + def __init__(self, EventLoop loop, Restorer restorer, FdAsyncIoStream stream): self.loop = loop self.restorer = restorer - self.network = _TwoPartyVatNetwork()._init(loop, deref(moveOwnAsyncIOStream(pipe.thisptr.ends[1])), capnp.SERVER) + self.network = _TwoPartyVatNetwork()._init(loop, deref(stream.thisptr), 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 +cdef class FdAsyncIoStream: + cdef Own[AsyncIoStream] thisptr - def __init__(self): - self.thisptr = newTwoWayPipe() + def __init__(self, int fd): + self.thisptr = AsyncIoStream_wrapFd(fd) cdef class _Schema: cdef C_Schema thisptr diff --git a/capnp/capnp_cpp.pxd b/capnp/capnp_cpp.pxd index e37ea5b..c799849 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, MessageBuilder +from schema_cpp cimport Node, Data, StructNode, EnumNode, MessageBuilder, MessageReader from async_cpp cimport PyPromise, VoidPromise, Promise from cpython.ref cimport PyObject @@ -23,6 +23,10 @@ cdef extern from "kj/string.h" namespace " ::kj": cdef cppclass String: char* cStr() +cdef extern from "kj/memory.h" namespace " ::kj": + cdef cppclass Own[T]: + T& operator*() + cdef extern from "kj/string-tree.h" namespace " ::kj": cdef cppclass StringTree: String flatten() @@ -42,14 +46,10 @@ cdef extern from "kj/array.h" namespace " ::kj": 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() + + Own[AsyncIoStream] AsyncIoStream_wrapFd" ::kj::AsyncIoStream::wrapFd"(int) cdef extern from "capnp/schema.h" namespace " ::capnp": cdef cppclass Schema: @@ -224,6 +224,7 @@ cdef extern from "rpcHelper.h": cdef cppclass PyRestorer: PyRestorer(PyObject *, StructSchema&) Capability.Client restoreHelper(RpcSystem&, MessageBuilder&) + Capability.Client restoreHelper(RpcSystem&, MessageReader&) cdef extern from "capnp/dynamic.h" namespace " ::capnp": cdef cppclass DynamicEnum: diff --git a/capnp/rpcHelper.h b/capnp/rpcHelper.h index 1dc18ea..56e5aa1 100644 --- a/capnp/rpcHelper.h +++ b/capnp/rpcHelper.h @@ -40,3 +40,10 @@ capnp::Capability::Client restoreHelper(capnp::RpcSystem()); } + + +capnp::Capability::Client restoreHelper(capnp::RpcSystem& client, capnp::MessageReader & objectId) { capnp::MallocMessageBuilder hostIdMessage(8); + auto hostId = hostIdMessage.initRoot(); + hostId.setSide(capnp::rpc::twoparty::Side::SERVER); + return client.restore(hostId, objectId.getRoot()); +} diff --git a/examples/example_capability.py b/examples/example_capability.py index 6812846..ba356e8 100644 --- a/examples/example_capability.py +++ b/examples/example_capability.py @@ -1,7 +1,8 @@ from __future__ import print_function import capnp -import example_capability_capnp +import example_capability_capnp as capability +import socket class Server: def __init__(self, val=1): @@ -10,24 +11,27 @@ class Server: def foo(self, context): context.results.x = str(context.params.i * 5 + self.val) -def test_simple_rpc(): +def example_simple_rpc(): def _restore(ref_id): - return example_capability_capnp.TestInterface.new_server(Server(100)) + return capability.TestInterface.new_server(Server(100)) loop = capnp.EventLoop() - pipe = capnp.TwoWayPipe() - restorer = capnp.Restorer(example_capability_capnp.TestSturdyRefObjectId, _restore) - server = capnp.RpcServer(loop, restorer, pipe) - client = capnp.RpcClient(loop, pipe) + read, write = socket.socketpair(socket.AF_UNIX) + read_stream = capnp.FdAsyncIoStream(read.fileno()) + write_stream = capnp.FdAsyncIoStream(write.fileno()) - ref = example_capability_capnp.TestSturdyRefObjectId.new_message() + restorer = capnp.Restorer(capability.TestSturdyRefObjectId, _restore) + server = capnp.RpcServer(loop, restorer, write_stream) + client = capnp.RpcClient(loop, read_stream) + + ref = capability.TestSturdyRefObjectId.new_message() cap = client.restore(ref.as_reader()) - cap = cap.cast_as(example_capability_capnp.TestInterface) + cap = cap.cast_as(capability.TestInterface) remote = cap.foo(i=5) response = loop.wait_remote(remote) assert response.x == '125' -test_simple_rpc() \ No newline at end of file +example_simple_rpc() \ No newline at end of file diff --git a/test/test_rpc.py b/test/test_rpc.py index c920ff5..ec29984 100644 --- a/test/test_rpc.py +++ b/test/test_rpc.py @@ -1,6 +1,7 @@ import pytest import capnp import os +import socket this_dir = os.path.dirname(__file__) @@ -20,14 +21,17 @@ def test_simple_rpc(capability): return capability.TestInterface.new_server(Server(100)) loop = capnp.EventLoop() - pipe = capnp.TwoWayPipe() - + + 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, restorer, pipe) - client = capnp.RpcClient(loop, pipe) + server = capnp.RpcServer(loop, restorer, write_stream) + client = capnp.RpcClient(loop, read_stream) ref = capability.TestSturdyRefObjectId.new_message() - cap = client.restore(ref.as_reader()) + cap = client.restore(ref) cap = cap.cast_as(capability.TestInterface) remote = cap.foo(i=5)