IQ.Lvbs/pycapnp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix #181

Browse Source
This commit is contained in:
Andrey Cizov
2019-07-13 14:08:48 +01:00
committed by Jacob Alexander
parent 0f0b30c666
commit 7bce2f751a
3 changed files with 15 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
capnp/helpers/helpers.pxd
View File

@@ -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 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": cdef extern from "capnp/helpers/rpcHelper.h":
Capability.Client bootstrapHelper(RpcSystem&) Capability.Client bootstrapHelper(RpcSystem&)
Capability.Client bootstrapHelperServer(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": cdef extern from "capnp/helpers/serialize.h":
ByteArray messageToPackedBytes(MessageBuilder &, size_t wordCount) ByteArray messageToPackedBytes(MessageBuilder &, size_t wordCount)

15
capnp/helpers/rpcHelper.h
View File

@@ -26,26 +26,25 @@ class ErrorHandler : public kj::TaskSet::ErrorHandler {
} }
}; };
struct ServerContext { struct ServerContext {
kj::Own<kj::AsyncIoStream> stream; kj::Own<kj::AsyncIoStream> stream;
capnp::TwoPartyVatNetwork network; capnp::TwoPartyVatNetwork network;
capnp::RpcSystem<capnp::rpc::twoparty::SturdyRefHostId> rpcSystem; capnp::RpcSystem<capnp::rpc::twoparty::SturdyRefHostId> rpcSystem;
ServerContext(kj::Own<kj::AsyncIoStream>&& stream, capnp::Capability::Client client) ServerContext(kj::Own<kj::AsyncIoStream>&& stream, capnp::Capability::Client client, capnp::ReaderOptions & opts)
: stream(kj::mv(stream)), : 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)) {} rpcSystem(makeRpcServer(network, client)) {}
}; };
void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own<kj::ConnectionReceiver>&& listener) { void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own<kj::ConnectionReceiver>&& listener, capnp::ReaderOptions & opts) {
auto ptr = listener.get(); auto ptr = listener.get();
tasks.add(ptr->accept().then(kj::mvCapture(kj::mv(listener), tasks.add(ptr->accept().then(kj::mvCapture(kj::mv(listener),
[&, client](kj::Own<kj::ConnectionReceiver>&& listener, [&, client](kj::Own<kj::ConnectionReceiver>&& listener,
kj::Own<kj::AsyncIoStream>&& connection) mutable { kj::Own<kj::AsyncIoStream>&& connection) mutable {
acceptLoop(tasks, client, kj::mv(listener)); acceptLoop(tasks, client, kj::mv(listener), opts);
auto server = kj::heap<ServerContext>(kj::mv(connection), client); auto server = kj::heap<ServerContext>(kj::mv(connection), client, opts);
// Arrange to destroy the server context when all references are gone, or when the // Arrange to destroy the server context when all references are gone, or when the
// EzRpcServer is destroyed (which will destroy the TaskSet). // EzRpcServer is destroyed (which will destroy the TaskSet).
@@ -53,7 +52,7 @@ void acceptLoop(kj::TaskSet & tasks, capnp::Capability::Client client, kj::Own<k
}))); })));
} }
kj::Promise<PyObject *> connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoContext * context, kj::StringPtr bindAddress) { kj::Promise<PyObject *> connectServer(kj::TaskSet & tasks, capnp::Capability::Client client, kj::AsyncIoContext * context, kj::StringPtr bindAddress, capnp::ReaderOptions & opts) {
auto paf = kj::newPromiseAndFulfiller<unsigned int>(); auto paf = kj::newPromiseAndFulfiller<unsigned int>();
auto portPromise = paf.promise.fork(); auto portPromise = paf.promise.fork();
@@ -63,7 +62,7 @@ kj::Promise<PyObject *> connectServer(kj::TaskSet & tasks, capnp::Capability::Cl
kj::Own<kj::NetworkAddress>&& addr) mutable { kj::Own<kj::NetworkAddress>&& addr) mutable {
auto listener = addr->listen(); auto listener = addr->listen();
portFulfiller->fulfill(listener->getPort()); 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); }); return portPromise.addBranch().then([&](unsigned int port) { return PyLong_FromUnsignedLong(port); });

12
capnp/lib/capnp.pyx
View File

@@ -2328,21 +2328,20 @@ cdef class TwoPartyServer:
raise KjException("You must provide a bootstrap interface to a server constructor.") raise KjException("You must provide a bootstrap interface to a server constructor.")
cdef _InterfaceSchema schema cdef _InterfaceSchema schema
cdef schema_cpp.ReaderOptions opts = make_reader_opts(traversal_limit_in_words, nesting_limit)
self._bootstrap = None self._bootstrap = None
if isinstance(socket, basestring): if isinstance(socket, basestring):
self._connect(socket, bootstrap) self._connect(socket, bootstrap, traversal_limit_in_words, nesting_limit)
return return
self._orig_stream = socket self._orig_stream = socket
if self._orig_stream: if self._orig_stream:
self._stream = _FdAsyncIoStream(socket.fileno()) 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: else:
# Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods # Initialize TwoWayPipe, to use pipe() acquire other end of the pipe using read() and write() methods
self._pipe = _TwoWayPipe() 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._server_socket = server_socket
self._port = 0 self._port = 0
@@ -2381,7 +2380,8 @@ cdef class TwoPartyServer:
len(data) len(data)
).wait(self._pipe._event_loop.thisptr.waitScope) ).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 _InterfaceSchema schema
cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER() cdef _EventLoop loop = C_DEFAULT_EVENT_LOOP_GETTER()
cdef capnp.StringPtr temp_string = capnp.StringPtr(<char*>host_string, len(host_string)) cdef capnp.StringPtr temp_string = capnp.StringPtr(<char*>host_string, len(host_string))
@@ -2390,7 +2390,7 @@ cdef class TwoPartyServer:
self._bootstrap = bootstrap self._bootstrap = bootstrap
Py_INCREF(self._bootstrap) Py_INCREF(self._bootstrap)
schema = bootstrap.schema schema = bootstrap.schema
self.port_promise = Promise()._init(helpers.connectServer(deref(self._task_set), helpers.server_to_client(schema.thisptr, <PyObject *>bootstrap), loop.thisptr, temp_string)) self.port_promise = Promise()._init(helpers.connectServer(deref(self._task_set), helpers.server_to_client(schema.thisptr, <PyObject *>bootstrap), loop.thisptr, temp_string, opts))
def _decref(self): def _decref(self):
Py_DECREF(self._bootstrap) Py_DECREF(self._bootstrap)