Struggling to understand AsyncIO example from "getting_started"

[rebcabin]

(editing in progress -- Post was saved too early by my incorrect button push. I'll say when editing is done)

I'm reading the example in the "IO Concurrency" section of the "getting started" guide for RxPy 4.0.4, here

https://rxpy.readthedocs.io/en/latest/get_started.html

The example runs and makes a loose kind of sense, but I can't make sense of the plumbing details, and thus will not be able to generalize the technique and use it creatively.

The code, as written, does not correspond to realities revealed in the debugger. I'll first say what I think the code means, from reading it, then I'll say what I uncovered by stepping through the running code.

from collections import namedtuple
import asyncio
import reactivex
import reactivex.operators as ops
from reactivex.subject import Subject
from reactivex.scheduler.eventloop import AsyncIOScheduler

EchoItem = namedtuple('EchoItem', ['future', 'data'])


def tcp_server(sink, loop):
    def on_subscribe(observer, scheduler):
        async def handle_echo(reader, writer):
            print("new client connected")
            while True:
                data = await reader.readline()
                data = data.decode("utf-8")
                if not data:
                    break

                future = asyncio.Future()
                observer.on_next(EchoItem(
                    future=future,
                    data=data
                ))
                await future
                writer.write(future.result().encode("utf-8"))

            print("Close the client socket")
            writer.close()

        def on_next(i):
            i.future.set_result(i.data)

        print("starting server")
        server = asyncio.start_server(handle_echo, '127.0.0.1', 8888)
        loop.create_task(server)

        sink.subscribe(
            on_next=on_next,
            on_error=observer.on_error,
            on_completed=observer.on_completed)

    return reactivex.create(on_subscribe)

[rebcabin]

I made a hash of this in the interactive editor. I'll make a new discussion.