diff --git a/examples/thread.capnp b/examples/thread.capnp new file mode 100644 index 0000000..ae32b8d --- /dev/null +++ b/examples/thread.capnp @@ -0,0 +1,11 @@ +@0xf5745ea9c82baa3a; + +interface Example { + interface StatusSubscriber { + status @0 (value: Bool); + # Call the function on the given parameters. + } + + longRunning @0 () -> (value: Bool); + subscribeStatus @1 (subscriber: StatusSubscriber); +} diff --git a/examples/thread_client.py b/examples/thread_client.py new file mode 100755 index 0000000..7201dd0 --- /dev/null +++ b/examples/thread_client.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python + +from __future__ import print_function + +import argparse +import threading +import time +import capnp + +import thread_capnp + +capnp.remove_event_loop() +capnp.create_event_loop(threaded=True) + + +def parse_args(): + parser = argparse.ArgumentParser(usage='Connects to the Example thread server \ +at the given address and does some RPCs') + parser.add_argument("host", help="HOST:PORT") + + return parser.parse_args() + + +class StatusSubscriber(thread_capnp.Example.StatusSubscriber.Server): + + '''An implementation of the StatusSubscriber interface''' + + def status(self, value, **kwargs): + print('status: {}'.format(time.time())) + + +def start_status_thread(host): + client = capnp.TwoPartyClient(host) + cap = client.ez_restore('example').cast_as(thread_capnp.Example) + + subscriber = StatusSubscriber() + promise = cap.subscribeStatus(subscriber) + promise.wait() + + +def main(host): + client = capnp.TwoPartyClient(host) + cap = client.ez_restore('example').cast_as(thread_capnp.Example) + + status_thread = threading.Thread(target=start_status_thread, args=(host,)) + status_thread.daemon = True + status_thread.start() + + print('main: {}'.format(time.time())) + cap.longRunning().wait() + print('main: {}'.format(time.time())) + cap.longRunning().wait() + print('main: {}'.format(time.time())) + cap.longRunning().wait() + print('main: {}'.format(time.time())) + +if __name__ == '__main__': + main(parse_args().host) diff --git a/examples/thread_server.py b/examples/thread_server.py new file mode 100755 index 0000000..04b1e2d --- /dev/null +++ b/examples/thread_server.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python + +from __future__ import print_function + +import argparse +import capnp + +import thread_capnp + + +class ExampleImpl(thread_capnp.Example.Server): + + "Implementation of the Example threading Cap'n Proto interface." + + def subscribeStatus(self, subscriber, **kwargs): + return capnp.getTimer().after_delay(10**9) \ + .then(lambda: subscriber.status(True)) \ + .then(lambda _: self.subscribeStatus(subscriber)) + + def longRunning(self, **kwargs): + return capnp.getTimer().after_delay(3 * 10**9) + + +def parse_args(): + parser = argparse.ArgumentParser(usage='''Runs the server bound to the\ +given address/port ADDRESS may be '*' to bind to all local addresses.\ +:PORT may be omitted to choose a port automatically. ''') + + parser.add_argument("address", help="ADDRESS[:PORT]") + + return parser.parse_args() + + +impl = ExampleImpl() + + +def restore(ref): + assert ref.as_text() == 'example' + return impl + + +def main(): + address = parse_args().address + + server = capnp.TwoPartyServer(address, restore) + server.run_forever() + +if __name__ == '__main__': + main()