From c3354e90693dc2e65db982c0138d951921cbaa0b Mon Sep 17 00:00:00 2001 From: Jason Paryani Date: Sat, 19 Oct 2013 22:38:10 -0700 Subject: [PATCH] Add ability to pipeline rpc requests --- capnp/capabilityHelper.h | 3 + capnp/capnp.pyx | 104 ++++++++++++++++++++++++++++++++- capnp/capnp_cpp.pxd | 8 ++- examples/example_capability.py | 2 +- test/test_capability.py | 17 +++++- 5 files changed, 127 insertions(+), 7 deletions(-) diff --git a/capnp/capabilityHelper.h b/capnp/capabilityHelper.h index eee76de..4bb9e40 100644 --- a/capnp/capabilityHelper.h +++ b/capnp/capabilityHelper.h @@ -73,6 +73,9 @@ public: capnp::DynamicCapability::Client new_client(capnp::InterfaceSchema & schema, PyObject * server, kj::EventLoop & loop) { return capnp::DynamicCapability::Client(kj::heap(schema, server), loop); } +capnp::DynamicValue::Reader new_server(capnp::InterfaceSchema & schema, PyObject * server) { + return capnp::DynamicValue::Reader(kj::heap(schema, server)); +} ::kj::Promise convert_to_pypromise(capnp::RemotePromise & promise) { return promise.then([](capnp::Response&& response) { return wrap_dynamic_struct_reader(response); } ); diff --git a/capnp/capnp.pyx b/capnp/capnp.pyx index fd4b647..48ae123 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, WordArrayPtr, DynamicCapability as C_DynamicCapability, new_client, 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, WordArrayPtr, DynamicCapability as C_DynamicCapability, new_client, new_server, Request, Response, RemotePromise, convert_to_pypromise, SimpleEventLoop, PyPromise, VoidPromise, CallContext from schema_cpp cimport Node as C_Node, EnumNode as C_EnumNode from cython.operator cimport dereference as deref @@ -113,6 +113,7 @@ cdef extern from "capnp/list.h" namespace " ::capnp": uint size() cdef extern from "" namespace "std": + C_DynamicStruct.Pipeline moveStructPipeline"std::move"(C_DynamicStruct.Pipeline) C_DynamicOrphan moveOrphan"std::move"(C_DynamicOrphan) Request moveRequest"std::move"(Request) Response moveResponse"std::move"(Response) @@ -375,6 +376,17 @@ cdef class _List_NestedNode_Reader: def __len__(self): return self.thisptr.size() +# cdef to_python_pipeline(C_DynamicValue.Pipeline self, object parent): +# cdef int type = self.getType() +# if type == capnp.TYPE_CAPABILITY: +# return _DynamicCapabilityClient()._init(self.asCapability(), parent) +# # elif type == capnp.TYPE_STRUCT: +# # return _DynamicStructReader()._init(self.asStruct(), parent) +# elif type == capnp.TYPE_UNKNOWN: +# raise ValueError("Cannot convert type to Python. Type is unknown by capnproto library") +# else: +# raise ValueError("Cannot convert type to Python. Type is unhandled by capnproto library") + cdef to_python_reader(C_DynamicValue.Reader self, object parent): cdef int type = self.getType() if type == capnp.TYPE_BOOL: @@ -448,6 +460,9 @@ cdef C_DynamicValue.Reader _extract_dynamic_struct_reader(_DynamicStructReader v cdef C_DynamicValue.Reader _extract_dynamic_client(_DynamicCapabilityClient value): return C_DynamicValue.Reader(value.thisptr) +cdef C_DynamicValue.Reader _extract_dynamic_server(_DynamicCapabilityServer value): + return new_server(value.schema.thisptr, value.server) + cdef _setDynamicField(_DynamicSetterClasses thisptr, field, value, parent): cdef C_DynamicValue.Reader temp value_type = type(value) @@ -480,6 +495,8 @@ cdef _setDynamicField(_DynamicSetterClasses thisptr, field, value, parent): thisptr.set(field, _extract_dynamic_struct_reader(value)) elif value_type is _DynamicCapabilityClient: thisptr.set(field, _extract_dynamic_client(value)) + elif value_type is _DynamicCapabilityServer: + thisptr.set(field, _extract_dynamic_server(value)) else: raise ValueError("Non primitive type") @@ -836,6 +853,53 @@ cdef class _DynamicStructBuilder: def to_dict(self): return _to_dict(self) +cdef class _DynamicStructPipeline: + """Reads Cap'n Proto structs + + This class is almost a 1 for 1 wrapping of the Cap'n Proto C++ DynamicStruct::Pipeline. The only difference is that instead of a `get` method, __getattr__ is overloaded and the field name is passed onto the C++ equivalent `get`. This means you just use . syntax to access any field. For field names that don't follow valid python naming convention for fields, use the global function :py:func:`getattr`:: + """ + cdef C_DynamicStruct.Pipeline * thisptr + cdef public object _parent + + cdef _init(self, C_DynamicStruct.Pipeline * other, object parent): + self.thisptr = other + self._parent = parent + return self + + def __dealloc__(self): + del self.thisptr + + cpdef _get(self, field) except +ValueError: + cdef int type = (self.thisptr.get(field)).getType() + if type == capnp.TYPE_CAPABILITY: + return _DynamicCapabilityClient()._init((self.thisptr.get(field)).asCapability(), self._parent) + elif type == capnp.TYPE_STRUCT: + return _DynamicStructPipeline()._init(new C_DynamicStruct.Pipeline(moveStructPipeline((self.thisptr.get(field)).asStruct())), self._parent) + elif type == capnp.TYPE_UNKNOWN: + raise ValueError("Cannot convert type to Python. Type is unknown by capnproto library") + else: + raise ValueError("Cannot convert type to Python. Type is unhandled by capnproto library") + + def __getattr__(self, field): + return self._get(field) + + property schema: + """A property that returns the _StructSchema object matching this reader""" + def __get__(self): + return _StructSchema()._init(self.thisptr.getSchema()) + + def __dir__(self): + return list(self.schema.fieldnames) + + # def __str__(self): + # return printStructReader(self.thisptr).flatten().cStr() + + # def __repr__(self): + # return '<%s reader %s>' % (self.schema.node.displayName, strStructReader(self.thisptr).cStr()) + + def to_dict(self): + return _to_dict(self) + cdef class _DynamicOrphan: cdef C_DynamicOrphan thisptr cdef public object _parent @@ -1024,7 +1088,15 @@ cdef class _RemotePromise: return _VoidPromise()._init(capnp.then(deref(self.thisptr), func, error_func)) cpdef _get(self, field) except +ValueError: - return _DynamicCapabilityClient()._init((self.thisptr.get(field)).asCapability(), self._parent) + cdef int type = (self.thisptr.get(field)).getType() + if type == capnp.TYPE_CAPABILITY: + return _DynamicCapabilityClient()._init((self.thisptr.get(field)).asCapability(), self._parent) + elif type == capnp.TYPE_STRUCT: + return _DynamicStructPipeline()._init(new C_DynamicStruct.Pipeline(moveStructPipeline((self.thisptr.get(field)).asStruct())), self._parent) + elif type == capnp.TYPE_UNKNOWN: + raise ValueError("Cannot convert type to Python. Type is unknown by capnproto library") + else: + raise ValueError("Cannot convert type to Python. Type is unhandled by capnproto library") def __getattr__(self, field): return self._get(field) @@ -1037,6 +1109,15 @@ cdef class _RemotePromise: def __dir__(self): return list(self.schema.fieldnames) + # def __str__(self): + # return printStructReader(self.thisptr).flatten().cStr() + + # def __repr__(self): + # return '<%s reader %s>' % (self.schema.node.displayName, strStructReader(self.thisptr).cStr()) + + def to_dict(self): + return _to_dict(self) + cdef class EventLoop: cdef SimpleEventLoop thisptr cpdef evalLater(self, func): @@ -1084,6 +1165,20 @@ cdef class _Response(_DynamicStructReader): self._init(deref(self.thisptr_child), parent) return self +cdef class _DynamicCapabilityServer: + cdef public _InterfaceSchema schema + cdef public object server + + def __init__(self, schema, server): + cdef _InterfaceSchema s + if hasattr(schema, 'schema'): + s = schema.schema + else: + s = schema + + self.schema = s + self.server = server + cdef class _DynamicCapabilityClient: cdef C_DynamicCapability.Client thisptr cdef public object _event_loop, _server, _parent @@ -1371,8 +1466,13 @@ cdef class SchemaParser: def helper(server, loop): return _DynamicCapabilityClient()._init_vals(bound_local_module, server, loop) return helper + def new_server(bound_local_module): + def helper(server): + return _DynamicCapabilityServer(bound_local_module, server) + return helper local_module.schema = schema.as_interface() local_module.new_client = new_client(local_module) + local_module.new_server = new_server(local_module) _load(schema, local_module) if not _os.path.isfile(file_name): diff --git a/capnp/capnp_cpp.pxd b/capnp/capnp_cpp.pxd index 077d2d5..33b8a9f 100644 --- a/capnp/capnp_cpp.pxd +++ b/capnp/capnp_cpp.pxd @@ -150,6 +150,8 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": DynamicOrphan disown(char *) DynamicStruct.Reader asReader() cppclass Pipeline: + Pipeline() + Pipeline(Pipeline &) DynamicValueForward.Pipeline get(char *) StructSchema getSchema() @@ -199,8 +201,9 @@ cdef extern from "capabilityHelper.h": PyPromise then(PyPromise & promise, PyObject * func, PyObject * error_func) VoidPromise then(RemotePromise & promise, PyObject * func, PyObject * error_func) cppclass PythonInterfaceDynamicImpl: - pass + PythonInterfaceDynamicImpl(PyObject *) DynamicCapability.Client new_client(InterfaceSchema&, PyObject *, EventLoop&) + DynamicValueForward.Reader new_server(InterfaceSchema&, PyObject *) PyPromise convert_to_pypromise(RemotePromise&) cdef extern from "capnp/dynamic.h" namespace " ::capnp": @@ -245,6 +248,7 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": Reader(DynamicEnum value) Reader(DynamicStruct.Reader& value) Reader(DynamicCapability.Client& value) + Reader(PythonInterfaceDynamicImpl& value) Type getType() int64_t asInt"as"() uint64_t asUint"as"() @@ -275,6 +279,8 @@ cdef extern from "capnp/dynamic.h" namespace " ::capnp": cppclass Pipeline: Pipeline(Pipeline) DynamicCapability.Client asCapability"releaseAs< ::capnp::DynamicCapability>"() + DynamicStruct.Pipeline asStruct"releaseAs< ::capnp::DynamicStruct>"() + Type getType() cdef extern from "capnp/schema-parser.h" namespace " ::capnp": cdef cppclass ParsedSchema(Schema): diff --git a/examples/example_capability.py b/examples/example_capability.py index 42c40c3..23059bd 100644 --- a/examples/example_capability.py +++ b/examples/example_capability.py @@ -12,7 +12,7 @@ def example_client(): client = example_capability_capnp.TestInterface.new_client(Server(), loop) - req = client.request('foo') + req = client._request('foo') req.i = 5 remote = req.send() diff --git a/test/test_capability.py b/test/test_capability.py index 696b713..cdad541 100644 --- a/test/test_capability.py +++ b/test/test_capability.py @@ -16,10 +16,13 @@ class Server: context.results.x = str(context.params.i * 5 + self.val) class PipelineServer: + def __init__(self, capability): + self.capability = capability + def getCap(self, context): def _then(response): context.results.s = response.x + '_foo' - context.results.outBox.outCap = Server(100) + context.results.outBox.cap = self.capability.TestInterface.new_server(Server(100)) return context.params.inCap.foo(i=context.params.n).then(_then) @@ -85,12 +88,20 @@ def test_simple_client(capability): def test_pipeline(capability): loop = capnp.EventLoop() - client = capability.TestPipeline.new_client(PipelineServer(), loop) + client = capability.TestPipeline.new_client(PipelineServer(capability), loop) foo_client = capability.TestInterface.new_client(Server(), loop) remote = client.getCap(n=5, inCap=foo_client) - response = loop.wait_remote(remote) + outCap = remote.outBox.cap + pipelinePromise = outCap.foo(i=10) + + response = loop.wait_remote(pipelinePromise) + assert response.x == '150' + + response = loop.wait_remote(remote) assert response.s == '26_foo' + +