Adding SSL version of the calculator example

- Used to test asyncio SSL on Windows mainly
- Skipping asyncio thread tests on Windows (due to getTimer wrapper bug
  on Windows)

examples/async_ssl_calculator_client.py

@@ -0,0 +1,353 @@
+#!/usr/bin/env python3
+
+import argparse
+import asyncio
+import os
+import socket
+import ssl
+
+import capnp
+import calculator_capnp
+
+
+this_dir = os.path.dirname(os.path.abspath(__file__))
+
+
+class PowerFunction(calculator_capnp.Calculator.Function.Server):
+
+    '''An implementation of the Function interface wrapping pow().  Note that
+    we're implementing this on the client side and will pass a reference to
+    the server.  The server will then be able to make calls back to the client.'''
+
+    def call(self, params, **kwargs):
+        '''Note the **kwargs. This is very necessary to include, since
+        protocols can add parameters over time. Also, by default, a _context
+        variable is passed to all server methods, but you can also return
+        results directly as python objects, and they'll be added to the
+        results struct in the correct order'''
+
+        return pow(params[0], params[1])
+
+
+async def myreader(client, reader):
+    while True:
+        data = await reader.read(4096)
+        client.write(data)
+
+
+async def mywriter(client, writer):
+    while True:
+        data = await client.read(4096)
+        writer.write(data.tobytes())
+        await writer.drain()
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(usage='Connects to the Calculator server \
+at the given address and does some RPCs')
+    parser.add_argument("host", help="HOST:PORT")
+
+    return parser.parse_args()
+
+
+async def main(host):
+    host = host.split(':')
+    addr = host[0]
+    port = host[1]
+
+    # Setup SSL context
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, 'selfsigned.cert'))
+
+    # Handle both IPv4 and IPv6 cases
+    try:
+        print("Try IPv4")
+        reader, writer = await asyncio.open_connection(
+            addr, port,
+            ssl=ctx,
+            family=socket.AF_INET
+        )
+    except Exception:
+        print("Try IPv6")
+        reader, writer = await asyncio.open_connection(
+            addr, port,
+            ssl=ctx,
+            family=socket.AF_INET6
+        )
+
+    # Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
+    client = capnp.TwoPartyClient()
+
+    # Assemble reader and writer tasks, run in the background
+    coroutines = [myreader(client, reader), mywriter(client, writer)]
+    asyncio.gather(*coroutines, return_exceptions=True)
+
+    # Bootstrap the Calculator interface
+    calculator = client.bootstrap().cast_as(calculator_capnp.Calculator)
+
+    '''Make a request that just evaluates the literal value 123.
+
+    What's interesting here is that evaluate() returns a "Value", which is
+    another interface and therefore points back to an object living on the
+    server.  We then have to call read() on that object to read it.
+    However, even though we are making two RPC's, this block executes in
+    *one* network round trip because of promise pipelining:  we do not wait
+    for the first call to complete before we send the second call to the
+    server.'''
+
+    print('Evaluating a literal... ', end="")
+
+    # Make the request. Note we are using the shorter function form (instead
+    # of evaluate_request), and we are passing a dictionary that represents a
+    # struct and its member to evaluate
+    eval_promise = calculator.evaluate({"literal": 123})
+
+    # This is equivalent to:
+    '''
+    request = calculator.evaluate_request()
+    request.expression.literal = 123
+
+    # Send it, which returns a promise for the result (without blocking).
+    eval_promise = request.send()
+    '''
+
+    # Using the promise, create a pipelined request to call read() on the
+    # returned object. Note that here we are using the shortened method call
+    # syntax read(), which is mostly just sugar for read_request().send()
+    read_promise = eval_promise.value.read()
+
+    # Now that we've sent all the requests, wait for the response.  Until this
+    # point, we haven't waited at all!
+    response = await read_promise.a_wait()
+    assert response.value == 123
+
+    print("PASS")
+
+    '''Make a request to evaluate 123 + 45 - 67.
+
+    The Calculator interface requires that we first call getOperator() to
+    get the addition and subtraction functions, then call evaluate() to use
+    them.  But, once again, we can get both functions, call evaluate(), and
+    then read() the result -- four RPCs -- in the time of *one* network
+    round trip, because of promise pipelining.'''
+
+    print("Using add and subtract... ", end='')
+
+    # Get the "add" function from the server.
+    add = calculator.getOperator(op='add').func
+    # Get the "subtract" function from the server.
+    subtract = calculator.getOperator(op='subtract').func
+
+    # Build the request to evaluate 123 + 45 - 67. Note the form is 'evaluate'
+    # + '_request', where 'evaluate' is the name of the method we want to call
+    request = calculator.evaluate_request()
+    subtract_call = request.expression.init('call')
+    subtract_call.function = subtract
+    subtract_params = subtract_call.init('params', 2)
+    subtract_params[1].literal = 67.0
+
+    add_call = subtract_params[0].init('call')
+    add_call.function = add
+    add_params = add_call.init('params', 2)
+    add_params[0].literal = 123
+    add_params[1].literal = 45
+
+    # Send the evaluate() request, read() the result, and wait for read() to finish.
+    eval_promise = request.send()
+    read_promise = eval_promise.value.read()
+
+    response = await read_promise.a_wait()
+    assert response.value == 101
+
+    print("PASS")
+
+    '''
+    Note: a one liner version of building the previous request (I highly
+    recommend not doing it this way for such a complicated structure, but I
+    just wanted to demonstrate it is possible to set all of the fields with a
+    dictionary):
+
+    eval_promise = calculator.evaluate(
+{'call': {'function': subtract,
+          'params': [{'call': {'function': add,
+                               'params': [{'literal': 123},
+                                          {'literal': 45}]}},
+                     {'literal': 67.0}]}})
+    '''
+
+    '''Make a request to evaluate 4 * 6, then use the result in two more
+    requests that add 3 and 5.
+
+    Since evaluate() returns its result wrapped in a `Value`, we can pass
+    that `Value` back to the server in subsequent requests before the first
+    `evaluate()` has actually returned.  Thus, this example again does only
+    one network round trip.'''
+
+    print("Pipelining eval() calls... ", end="")
+
+    # Get the "add" function from the server.
+    add = calculator.getOperator(op='add').func
+    # Get the "multiply" function from the server.
+    multiply = calculator.getOperator(op='multiply').func
+
+    # Build the request to evaluate 4 * 6
+    request = calculator.evaluate_request()
+
+    multiply_call = request.expression.init("call")
+    multiply_call.function = multiply
+    multiply_params = multiply_call.init("params", 2)
+    multiply_params[0].literal = 4
+    multiply_params[1].literal = 6
+
+    multiply_result = request.send().value
+
+    # Use the result in two calls that add 3 and add 5.
+
+    add_3_request = calculator.evaluate_request()
+    add_3_call = add_3_request.expression.init("call")
+    add_3_call.function = add
+    add_3_params = add_3_call.init("params", 2)
+    add_3_params[0].previousResult = multiply_result
+    add_3_params[1].literal = 3
+    add_3_promise = add_3_request.send().value.read()
+
+    add_5_request = calculator.evaluate_request()
+    add_5_call = add_5_request.expression.init("call")
+    add_5_call.function = add
+    add_5_params = add_5_call.init("params", 2)
+    add_5_params[0].previousResult = multiply_result
+    add_5_params[1].literal = 5
+    add_5_promise = add_5_request.send().value.read()
+
+    # Now wait for the results.
+    assert (await add_3_promise.a_wait()).value == 27
+    assert (await add_5_promise.a_wait()).value == 29
+
+    print("PASS")
+
+    '''Our calculator interface supports defining functions.  Here we use it
+    to define two functions and then make calls to them as follows:
+
+      f(x, y) = x * 100 + y
+      g(x) = f(x, x + 1) * 2;
+      f(12, 34)
+      g(21)
+
+    Once again, the whole thing takes only one network round trip.'''
+
+    print("Defining functions... ", end="")
+
+    # Get the "add" function from the server.
+    add = calculator.getOperator(op='add').func
+    # Get the "multiply" function from the server.
+    multiply = calculator.getOperator(op='multiply').func
+
+    # Define f.
+    request = calculator.defFunction_request()
+    request.paramCount = 2
+
+    # Build the function body.
+    add_call = request.body.init("call")
+    add_call.function = add
+    add_params = add_call.init("params", 2)
+    add_params[1].parameter = 1  # y
+
+    multiply_call = add_params[0].init("call")
+    multiply_call.function = multiply
+    multiply_params = multiply_call.init("params", 2)
+    multiply_params[0].parameter = 0  # x
+    multiply_params[1].literal = 100
+
+    f = request.send().func
+
+    # Define g.
+    request = calculator.defFunction_request()
+    request.paramCount = 1
+
+    # Build the function body.
+    multiply_call = request.body.init("call")
+    multiply_call.function = multiply
+    multiply_params = multiply_call.init("params", 2)
+    multiply_params[1].literal = 2
+
+    f_call = multiply_params[0].init("call")
+    f_call.function = f
+    f_params = f_call.init("params", 2)
+    f_params[0].parameter = 0
+
+    add_call = f_params[1].init("call")
+    add_call.function = add
+    add_params = add_call.init("params", 2)
+    add_params[0].parameter = 0
+    add_params[1].literal = 1
+
+    g = request.send().func
+
+    # OK, we've defined all our functions.  Now create our eval requests.
+
+    # f(12, 34)
+    f_eval_request = calculator.evaluate_request()
+    f_call = f_eval_request.expression.init("call")
+    f_call.function = f
+    f_params = f_call.init("params", 2)
+    f_params[0].literal = 12
+    f_params[1].literal = 34
+    f_eval_promise = f_eval_request.send().value.read()
+
+    # g(21)
+    g_eval_request = calculator.evaluate_request()
+    g_call = g_eval_request.expression.init("call")
+    g_call.function = g
+    g_call.init('params', 1)[0].literal = 21
+    g_eval_promise = g_eval_request.send().value.read()
+
+    # Wait for the results.
+    assert (await f_eval_promise.a_wait()).value == 1234
+    assert (await g_eval_promise.a_wait()).value == 4244
+
+    print("PASS")
+
+    '''Make a request that will call back to a function defined locally.
+
+    Specifically, we will compute 2^(4 + 5).  However, exponent is not
+    defined by the Calculator server.  So, we'll implement the Function
+    interface locally and pass it to the server for it to use when
+    evaluating the expression.
+
+    This example requires two network round trips to complete, because the
+    server calls back to the client once before finishing.  In this
+    particular case, this could potentially be optimized by using a tail
+    call on the server side -- see CallContext::tailCall().  However, to
+    keep the example simpler, we haven't implemented this optimization in
+    the sample server.'''
+
+    print("Using a callback... ", end="")
+
+    # Get the "add" function from the server.
+    add = calculator.getOperator(op='add').func
+
+    # Build the eval request for 2^(4+5).
+    request = calculator.evaluate_request()
+
+    pow_call = request.expression.init("call")
+    pow_call.function = PowerFunction()
+    pow_params = pow_call.init("params", 2)
+    pow_params[0].literal = 2
+
+    add_call = pow_params[1].init("call")
+    add_call.function = add
+    add_params = add_call.init("params", 2)
+    add_params[0].literal = 4
+    add_params[1].literal = 5
+
+    # Send the request and wait.
+    response = await request.send().value.read().a_wait()
+    assert response.value == 512
+
+    print("PASS")
+
+if __name__ == '__main__':
+    # Using asyncio.run hits an asyncio ssl bug
+    # https://bugs.python.org/issue36709
+    # asyncio.run(main(parse_args().host), loop=loop, debug=True)
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main(parse_args().host))

examples/async_ssl_calculator_server.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+
+import argparse
+import asyncio
+import logging
+import os
+import socket
+import ssl
+
+import capnp
+import calculator_capnp
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+this_dir = os.path.dirname(os.path.abspath(__file__))
+
+
+class Server:
+    async def myreader(self):
+        while self.retry:
+            try:
+                # Must be a wait_for so we don't block on read()
+                data = await asyncio.wait_for(
+                    self.reader.read(4096),
+                    timeout=0.1
+                )
+            except asyncio.TimeoutError:
+                logger.debug("myreader timeout.")
+                continue
+            except Exception as err:
+                logger.error("Unknown myreader err: %s", err)
+                return False
+            await self.server.write(data)
+        logger.debug("myreader done.")
+        return True
+
+
+    async def mywriter(self):
+        while self.retry:
+            try:
+                # Must be a wait_for so we don't block on read()
+                data = await asyncio.wait_for(
+                    self.server.read(4096),
+                    timeout=0.1
+                )
+                self.writer.write(data.tobytes())
+            except asyncio.TimeoutError:
+                logger.debug("mywriter timeout.")
+                continue
+            except Exception as err:
+                logger.error("Unknown mywriter err: %s", err)
+                return False
+        logger.debug("mywriter done.")
+        return True
+
+
+    async def myserver(self, reader, writer):
+        # Start TwoPartyServer using TwoWayPipe (only requires bootstrap)
+        self.server = capnp.TwoPartyServer(bootstrap=CalculatorImpl())
+        self.reader = reader
+        self.writer = writer
+        self.retry = True
+
+        # Assemble reader and writer tasks, run in the background
+        coroutines = [self.myreader(), self.mywriter()]
+        tasks = asyncio.gather(*coroutines, return_exceptions=True)
+
+        while True:
+            self.server.poll_once()
+            # Check to see if reader has been sent an eof (disconnect)
+            if self.reader.at_eof():
+                self.retry = False
+                break
+            await asyncio.sleep(0.01)
+
+        # Make wait for reader/writer to finish (prevent possible resource leaks)
+        await tasks
+
+
+def read_value(value):
+    '''Helper function to asynchronously call read() on a Calculator::Value and
+    return a promise for the result.  (In the future, the generated code might
+    include something like this automatically.)'''
+
+    return value.read().then(lambda result: result.value)
+
+
+def evaluate_impl(expression, params=None):
+    '''Implementation of CalculatorImpl::evaluate(), also shared by
+    FunctionImpl::call().  In the latter case, `params` are the parameter
+    values passed to the function; in the former case, `params` is just an
+    empty list.'''
+
+    which = expression.which()
+
+    if which == 'literal':
+        return capnp.Promise(expression.literal)
+    elif which == 'previousResult':
+        return read_value(expression.previousResult)
+    elif which == 'parameter':
+        assert expression.parameter < len(params)
+        return capnp.Promise(params[expression.parameter])
+    elif which == 'call':
+        call = expression.call
+        func = call.function
+
+        # Evaluate each parameter.
+        paramPromises = [evaluate_impl(param, params) for param in call.params]
+
+        joinedParams = capnp.join_promises(paramPromises)
+        # When the parameters are complete, call the function.
+        ret = (joinedParams
+               .then(lambda vals: func.call(vals))
+               .then(lambda result: result.value))
+
+        return ret
+    else:
+        raise ValueError("Unknown expression type: " + which)
+
+
+class ValueImpl(calculator_capnp.Calculator.Value.Server):
+
+    "Simple implementation of the Calculator.Value Cap'n Proto interface."
+
+    def __init__(self, value):
+        self.value = value
+
+    def read(self, **kwargs):
+        return self.value
+
+
+class FunctionImpl(calculator_capnp.Calculator.Function.Server):
+
+    '''Implementation of the Calculator.Function Cap'n Proto interface, where the
+    function is defined by a Calculator.Expression.'''
+
+    def __init__(self, paramCount, body):
+        self.paramCount = paramCount
+        self.body = body.as_builder()
+
+    def call(self, params, _context, **kwargs):
+        '''Note that we're returning a Promise object here, and bypassing the
+        helper functionality that normally sets the results struct from the
+        returned object. Instead, we set _context.results directly inside of
+        another promise'''
+
+        assert len(params) == self.paramCount
+        # using setattr because '=' is not allowed inside of lambdas
+        return evaluate_impl(self.body, params).then(lambda value: setattr(_context.results, 'value', value))
+
+
+class OperatorImpl(calculator_capnp.Calculator.Function.Server):
+
+    '''Implementation of the Calculator.Function Cap'n Proto interface, wrapping
+    basic binary arithmetic operators.'''
+
+    def __init__(self, op):
+        self.op = op
+
+    def call(self, params, **kwargs):
+        assert len(params) == 2
+
+        op = self.op
+
+        if op == 'add':
+            return params[0] + params[1]
+        elif op == 'subtract':
+            return params[0] - params[1]
+        elif op == 'multiply':
+            return params[0] * params[1]
+        elif op == 'divide':
+            return params[0] / params[1]
+        else:
+            raise ValueError('Unknown operator')
+
+
+class CalculatorImpl(calculator_capnp.Calculator.Server):
+
+    "Implementation of the Calculator Cap'n Proto interface."
+
+    def evaluate(self, expression, _context, **kwargs):
+        return evaluate_impl(expression).then(lambda value: setattr(_context.results, 'value', ValueImpl(value)))
+
+    def defFunction(self, paramCount, body, _context, **kwargs):
+        return FunctionImpl(paramCount, body)
+
+    def getOperator(self, op, **kwargs):
+        return OperatorImpl(op)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(usage='''Runs the server bound to the\
+given address/port ADDRESS. ''')
+
+    parser.add_argument("address", help="ADDRESS:PORT")
+
+    return parser.parse_args()
+
+
+async def new_connection(reader, writer):
+    server = Server()
+    await server.myserver(reader, writer)
+
+
+async def main():
+    address = parse_args().address
+    host = address.split(':')
+    addr = host[0]
+    port = host[1]
+
+    # Setup SSL context
+    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    ctx.load_cert_chain(os.path.join(this_dir, 'selfsigned.cert'), os.path.join(this_dir, 'selfsigned.key'))
+
+    # Handle both IPv4 and IPv6 cases
+    try:
+        print("Try IPv4")
+        server = await asyncio.start_server(
+            new_connection,
+            addr, port,
+            ssl=ctx,
+            family=socket.AF_INET
+        )
+    except Exception:
+        print("Try IPv6")
+        server = await asyncio.start_server(
+            new_connection,
+            addr, port,
+            ssl=ctx,
+            family=socket.AF_INET6
+        )
+
+    async with server:
+        await server.serve_forever()
+
+if __name__ == '__main__':
+    asyncio.run(main())

examples/async_ssl_client.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
 
-import asyncio
 import argparse
+import asyncio
 import os
-import time
-import capnp
 import socket
 import ssl
+import time
 
+import capnp
 import thread_capnp
 
 this_dir = os.path.dirname(os.path.abspath(__file__))

examples/async_ssl_server.py

@@ -11,7 +11,7 @@ import capnp
 import thread_capnp
 
 
-logger = logging.getLogger(__)
+logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 
 this_dir = os.path.dirname(os.path.abspath(__file__))

test/test_examples.py

@@ -126,6 +126,7 @@ def test_addressbook_example(cleanup):
     assert ret == 0
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196")
 def test_async_example(cleanup):
     address = '{}:36434'.format(hostname)
     server = 'async_server.py'
@@ -133,6 +134,7 @@ def test_async_example(cleanup):
     run_subprocesses(address, server, client)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196")
 def test_ssl_async_example(cleanup):
     address = '{}:36435'.format(hostname)
     server = 'async_ssl_server.py'
@@ -140,8 +142,16 @@ def test_ssl_async_example(cleanup):
     run_subprocesses(address, server, client)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196")
 def test_ssl_reconnecting_async_example(cleanup):
     address = '{}:36436'.format(hostname)
     server = 'async_ssl_server.py'
     client = 'async_reconnecting_ssl_client.py'
     run_subprocesses(address, server, client)
+
+
+def test_async_ssl_calculator_example(cleanup):
+    address = '{}:36437'.format(hostname)
+    server = 'async_ssl_calculator_server.py'
+    client = 'async_ssl_calculator_client.py'
+    run_subprocesses(address, server, client)