How to use it

The following example is an http echo server:

from collections import namedtuple

from cyclotron import Component
from cyclotron.asyncio.runner import run
import cyclotron_aiohttp.httpd as httpd
import reactivex as rx
import reactivex.operators as ops

EchoSource = namedtuple('EchoSource', ['httpd'])
EchoSink = namedtuple('EchoSink', ['httpd'])
EchoDrivers = namedtuple('EchoDrivers', ['httpd'])

def echo_server(source):
    init = rx.from_([
        httpd.Initialize(),
        httpd.AddRoute(methods=['GET'], path='/echo/{what}', id='echo'),
        httpd.StartServer(host='localhost', port=8080),
    ])

    echo = source.httpd.route.pipe(
        ops.filter(lambda i: i.id == 'echo'),
        ops.flat_map(lambda i: i.request),
        ops.map(lambda i: httpd.Response(
            context=i.context,
            data=i.match_info['what'].encode('utf-8')),
        )
    )

    control = rx.merge(init, echo)
    return EchoSink(httpd=httpd.Sink(control=control))


def main():
    run(Component(call=echo_server, input=EchoSource),
        EchoDrivers(httpd=httpd.make_driver()))


if __name__ == '__main__':
    main()

In this application, the echo_server function is a pure function, while the http server is implemented as a driver.

pip install cyclotron-aiohttp

you can then test it with an http client like curl:

$ curl http://localhost:8080/echo/hello
hello

Install

Cyclotron is available on PyPi and can be installed with pip:

pip install cyclotron

Cyclotron automatically uses uvloop if it is available.

This project is composed of several python packages. Install also the ones that you use in your application:

License

This project is licensed under the MIT License - see the License file for details