def crossroad(request, source, drain=False):
sink_observer = None<EOL>def on_sink_subscribe(observer):<EOL><INDENT>nonlocal sink_observer<EOL>sink_observer = observer<EOL><DEDENT>def on_response_subscribe(observer):<EOL><INDENT>def on_next_source(i):<EOL><INDENT>if type(i) is cyclotron.Drain:<EOL><INDENT>observer.on_completed()<EOL><DEDENT>else:<EOL><INDENT>observer.on_next(i)<EOL><DEDENT><DEDENT>source_disposable = source.subscribe(<EOL>on_next=on_next_source,<EOL>on_error=lambda e: observer.on_error(e),<EOL>on_completed=lambda: observer.on_completed()<EOL>)<EOL>def on_next_request(i):<EOL><INDENT>if sink_observer is not None:<EOL><INDENT>sink_observer.on_next(i)<EOL><DEDENT><DEDENT>def on_request_completed():<EOL><INDENT>if sink_observer is not None:<EOL><INDENT>if drain is True:<EOL><INDENT>sink_observer.on_next(cyclotron.Drain())<EOL><DEDENT>else:<EOL><INDENT>sink_observer.on_completed()<EOL><DEDENT><DEDENT><DEDENT>request_disposable = request.subscribe(<EOL>on_next=on_next_request,<EOL>on_error=observer.on_error,<EOL>on_completed=on_request_completed<EOL>)<EOL>def dispose():<EOL><INDENT>source_disposable.dispose()<EOL>request_disposable.dispose()<EOL><DEDENT>return dispose<EOL><DEDENT>return (<EOL>Observable.create(on_sink_subscribe),<EOL>Observable.create(on_response_subscribe)<EOL>)<EOL>
Creates a crossroad router

    A crossroad is a cross-routing between two pair of sink/source
    observables. This allows to use drivers with less discruption on
    the Observable chain. A crossroad has the following structure:

    .. image:: ../docs/asset/crossroad.png
        :scale: 60%
        :align: center

    This function is a lettable operator.

    Parameters
    ----------
    request : Observable
        The observable coming from the observable chain.

    source : Observable
        The source observable of the driver to wrap.

    drain : Boolean
        Indicated whether items should be drained form the driver before
        forwarding completion events.

    Returns
    -------
    sink : Observable
        A sink observable that must be routed to the sink observable of the
        driver.

    response : Observable
        The response coming from the driver.