Add FdAsyncIoStream. Also clean up RPC interface a bit

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 "<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])
+    Own[AsyncIoStream] moveOwnAsyncIOStream"std::move"(Own[AsyncIoStream])
 
 cdef extern from "<capnp/pretty-print.h>" 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+:
+    cpdef restore(self, objectId) except+:
-        cdef _MessageBuilder builder = objectId._parent
+        cdef _MessageBuilder builder 
-        return _CapabilityClient()._init(restoreHelper(deref(self.thisptr), deref(builder.thisptr)), self)
+        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 class FdAsyncIoStream:
-    cdef C_TwoWayPipe thisptr
+    cdef Own[AsyncIoStream] thisptr
 
-    def __init__(self):
+    def __init__(self, int fd):
-        self.thisptr = newTwoWayPipe()
+        self.thisptr = AsyncIoStream_wrapFd(fd)
 
 cdef class _Schema:
     cdef C_Schema thisptr

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
+    Own[AsyncIoStream] AsyncIoStream_wrapFd" ::kj::AsyncIoStream::wrapFd"(int)
-    TwoWayPipe newTwoWayPipe()
 
 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:

capnp/rpcHelper.h

@@ -40,3 +40,10 @@ capnp::Capability::Client restoreHelper(capnp::RpcSystem<capnp::rpc::twoparty::S
     hostId.setSide(capnp::rpc::twoparty::Side::SERVER);
     return client.restore(hostId, objectId.getRoot<capnp::ObjectPointer>());
 }
+
+
+capnp::Capability::Client restoreHelper(capnp::RpcSystem<capnp::rpc::twoparty::SturdyRefHostId>& client, capnp::MessageReader & objectId) {  capnp::MallocMessageBuilder hostIdMessage(8);
+    auto hostId = hostIdMessage.initRoot<capnp::rpc::twoparty::SturdyRefHostId>();
+    hostId.setSide(capnp::rpc::twoparty::Side::SERVER);
+    return client.restore(hostId, objectId.getRoot<capnp::ObjectPointer>());
+}

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()
+    read, write = socket.socketpair(socket.AF_UNIX)
-    restorer = capnp.Restorer(example_capability_capnp.TestSturdyRefObjectId, _restore)
+    read_stream = capnp.FdAsyncIoStream(read.fileno())
-    server = capnp.RpcServer(loop, restorer, pipe)
+    write_stream = capnp.FdAsyncIoStream(write.fileno())
-    client = capnp.RpcClient(loop, pipe)
 
-    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()
+example_simple_rpc()

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)
+    server = capnp.RpcServer(loop, restorer, write_stream)
-    client = capnp.RpcClient(loop, pipe)
+    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)