Initial wrapping of rpc-twoparty functionality

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

capnp/capabilityHelper.h

@@ -1,3 +1,5 @@
+#pragma once
+
 #include "capnp/dynamic.h"
 #include <stdexcept>
 #include "Python.h"
@@ -97,6 +99,10 @@ capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject
   return capnp::DynamicValue::Reader(kj::heap<PythonInterfaceDynamicImpl>(schema, server));
 }
 
+capnp::Capability::Client server_to_client(capnp::InterfaceSchema & schema, PyObject * server) {
+  return kj::heap<PythonInterfaceDynamicImpl>(schema, server);
+}
+
 ::kj::Promise<PyObject *> convert_to_pypromise(capnp::RemotePromise<capnp::DynamicStruct> & promise) {
     return promise.then([](capnp::Response<capnp::DynamicStruct>&& response) { return wrap_dynamic_struct_reader(response); } );
-}
+}

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 = <object>_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, <PyObject *>server.server))
+
 cdef public object wrap_kj_exception(capnp.Exception & exception):
     return None # TODO
 
@@ -128,6 +137,7 @@ cdef extern from "<utility>" 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 "<capnp/pretty-print.h>" 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, <PyObject *>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(<PyObject*>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):

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<capnp::rpc::twoparty::SturdyRefHostId>":
+        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
+

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;
+  }
+}

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')
+    pipe = capnp.TwoWayPipe()
-    req.i = 5
+    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__':
+test_simple_rpc()
-    example_client()

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;
+  }
+}