def make_crossroad_router(source, drain=False):
sink_observer = None<EOL>def on_sink_subscribe(observer):<EOL><INDENT>nonlocal sink_observer<EOL>sink_observer = observer<EOL>def dispose():<EOL><INDENT>nonlocal sink_observer<EOL>sink_observer = None<EOL><DEDENT>return dispose<EOL><DEDENT>def route_crossroad(request):<EOL><INDENT>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 Observable.create(on_response_subscribe)<EOL><DEDENT>return Observable.create(on_sink_subscribe), route_crossroad<EOL>
legacy crossroad implementation. deprecated