From 7bce2f751a2a60c430c9459fc1b04f1a3f2ef58d Mon Sep 17 00:00:00 2001 From: Andrey Cizov Date: Sat, 13 Jul 2019 14:08:48 +0100 Subject: [PATCH] fix #181 --- capnp/helpers/helpers.pxd | 4 ++-- capnp/helpers/rpcHelper.h | 15 +++++++-------- capnp/lib/capnp.pyx | 12 ++++++------ 3 files changed, 15 insertions(+), 16 deletions(-) diff --git a/capnp/helpers/helpers.pxd b/capnp/helpers/helpers.pxd index f34b530..1cc0a03 100644 --- a/capnp/helpers/helpers.pxd +++ b/capnp/helpers/helpers.pxd @@ -1,4 +1,4 @@ -from capnp.includes.capnp_cpp cimport Maybe, DynamicStruct, Request, Response, PyPromise, VoidPromise, PyPromiseArray, RemotePromise, DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, Capability, RpcSystem, MessageBuilder, MessageReader, TwoPartyVatNetwork, AnyPointer, DynamicStruct_Builder, WaitScope, AsyncIoContext, StringPtr, TaskSet, Timer +from capnp.includes.capnp_cpp cimport Maybe, ReaderOptions, DynamicStruct, Request, Response, PyPromise, VoidPromise, PyPromiseArray, RemotePromise, DynamicCapability, InterfaceSchema, EnumSchema, StructSchema, DynamicValue, Capability, RpcSystem, MessageBuilder, MessageReader, TwoPartyVatNetwork, AnyPointer, DynamicStruct_Builder, WaitScope, AsyncIoContext, StringPtr, TaskSet, Timer from capnp.includes.schema_cpp cimport ByteArray @@ -29,7 +29,7 @@ cdef extern from "capnp/helpers/capabilityHelper.h": cdef extern from "capnp/helpers/rpcHelper.h": Capability.Client bootstrapHelper(RpcSystem&) Capability.Client bootstrapHelperServer(RpcSystem&) - PyPromise connectServer(TaskSet &, Capability.Client, AsyncIoContext *, StringPtr) + PyPromise connectServer(TaskSet &, Capability.Client, AsyncIoContext *, StringPtr, ReaderOptions &) cdef extern from "capnp/helpers/serialize.h": ByteArray messageToPackedBytes(MessageBuilder &, size_t wordCount) diff --git a/capnp/helpers/rpcHelper.h b/capnp/helpers/rpcHelper.h index 44f098f..192e16a 100644 --- a/capnp/helpers/rpcHelper.h +++ b/capnp/helpers/rpcHelper.h @@ -26,26 +26,25 @@ class ErrorHandler : public kj::TaskSet::ErrorHandler { } }; - struct ServerContext { kj::Own stream; capnp::TwoPartyVatNetwork network; capnp::RpcSystem rpcSystem; - ServerContext(kj::Own&& stream, capnp::Capability::Client client) + ServerContext(kj::Own&& stream, capnp::Capability::Client client, capnp::ReaderOptions & opts) : stream(kj::mv(stream)), - network(*this->stream, capnp::rpc::twoparty::Side::SERVER), + network(*this->stream, capnp::rpc::twoparty::Side::SERVER, opts), rpcSystem(makeRpcServer(network, client)) {} }; -void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own&& listener) { +void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own&& listener, capnp::ReaderOptions & opts) { auto ptr = listener.get(); tasks.add(ptr->accept().then(kj::mvCapture(kj::mv(listener), [&, client](kj::Own&& listener, kj::Own&& connection) mutable { - acceptLoop(tasks, client, kj::mv(listener)); + acceptLoop(tasks, client, kj::mv(listener), opts); - auto server = kj::heap(kj::mv(connection), client); + auto server = kj::heap(kj::mv(connection), client, opts); // Arrange to destroy the server context when all references are gone, or when the // EzRpcServer is destroyed (which will destroy the TaskSet). @@ -53,7 +52,7 @@ void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoContext * context, kj::StringPtr bindAddress) { +kj::Promise connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoContext * context, kj::StringPtr bindAddress, capnp::ReaderOptions & opts) { auto paf = kj::newPromiseAndFulfiller(); auto portPromise = paf.promise.fork(); @@ -63,7 +62,7 @@ kj::Promise connectServer(kj::TaskSet & tasks, capnp::Capability::Cl kj::Own&& addr) mutable { auto listener = addr->listen(); portFulfiller->fulfill(listener->getPort()); - acceptLoop(tasks, client, kj::mv(listener)); + acceptLoop(tasks, client, kj::mv(listener), opts); }))); return portPromise.addBranch().then([&](unsigned int port) { return PyLong_FromUnsignedLong(port); }); diff --git a/capnp/lib/capnp.pyx b/capnp/lib/capnp.pyx index 28df235..dfcab60 100644 --- a/capnp/lib/capnp.pyx +++ b/capnp/lib/capnp.pyx @@ -2328,21 +2328,20 @@ cdef class TwoPartyServer: raise KjException("You must provide a bootstrap interface to a server constructor.") cdef _InterfaceSchema schema - cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit) self._bootstrap = None if isinstance(socket, basestring): - self._connect(socket, bootstrap) + self._connect(socket, bootstrap, traversal_limit_in_words, nesting_limit) return self._orig_stream = socket if self._orig_stream: self._stream = _FdAsyncIoStream(socket.fileno()) - self._network = _TwoPartyVatNetwork()._init(self._stream, capnp.SERVER, opts) + self._network = _TwoPartyVatNetwork()._init(self._stream, capnp.SERVER, make_reader_opts(traversal_limit_in_words, nesting_limit)) else: # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods self._pipe = _TwoWayPipe() - self._network = _TwoPartyVatNetwork()._init_pipe(self._pipe, capnp.SERVER, opts) + self._network = _TwoPartyVatNetwork()._init_pipe(self._pipe, capnp.SERVER, make_reader_opts(traversal_limit_in_words, nesting_limit)) self._server_socket = server_socket self._port = 0 @@ -2381,7 +2380,8 @@ cdef class TwoPartyServer: len(data) ).wait(self._pipe._event_loop.thisptr.waitScope) - cpdef _connect(self, host_string, bootstrap): + cpdef _connect(self, host_string, bootstrap, traversal_limit_in_words, nesting_limit): + cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit) cdef _InterfaceSchema schema cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() cdef capnp.StringPtr temp_string = capnp.StringPtr(host_string, len(host_string)) @@ -2390,7 +2390,7 @@ cdef class TwoPartyServer: self._bootstrap = bootstrap Py_INCREF(self._bootstrap) schema = bootstrap.schema - self.port_promise = Promise()._init(helpers.connectServer(deref(self._task_set), helpers.server_to_client(schema.thisptr, bootstrap), loop.thisptr, temp_string)) + self.port_promise = Promise()._init(helpers.connectServer(deref(self._task_set), helpers.server_to_client(schema.thisptr, bootstrap), loop.thisptr, temp_string, opts)) def _decref(self): Py_DECREF(self._bootstrap)