Creating a shared, asynchronous, cancellable Observable with a custom function in RxPy

Question

This is an exercise in using ReactiveX in Python. I want to create an Observable in RxPy with the following characteristics:

• Emits items through a custom function
• Runs asynchronously
• Sets up resources upon start and it cleans up at the end
• Runs a loop indefinitely (until cancelled)
• Can be cancelled
• Shared among observers

Kind of like wrapping HTTP streaming data with an Observable.

Thinking that it would work like RxJs, I started defining a straightforward function and using the create operator. It got hairy very quick and I ended up doing a lot of asyncio stuff that I wasn't expecting to do.

How does this code look and how could I improve it (particularly observable_fn)?

You can run it with Python 3.5 and stop it with ctrl+C. I'm working on a Mac.

import asyncio
from rx import Observer, Observable
from rx.core import Scheduler

def observable_fn(observer):
    _task = None

    # Customize this
    def _setup():
        print('Observable setup')

    # Customize this
    def _teardown():
        print('Observable teardown')
        if _task:
            _task.cancel()
        observer.on_completed()

    # Customize this
    async def _loop():
        counter = 0

        while True:
            await asyncio.sleep(2) # Simulating long-running task
            counter += 1
            observer.on_next(counter)

    # Don't touch this
    async def _run_loop():
        try:
            await _loop()
        except asyncio.CancelledError:
            print('Observable cancelled')
        finally:
            _teardown()

    # Don't touch this
    _setup()
    loop = asyncio.get_event_loop()
    _task = loop.create_task(_run_loop())

print('CREATING OBSERVABLE')
observable = Observable.create(observable_fn).observe_on(Scheduler.event_loop).share()

print('CREATING OBSERVERS')
observer1 = observable.subscribe(lambda x: print('O1: {0}'.format(x)))
observer2 = observable.subscribe(lambda x: print('O2: {0}'.format(x)))
observer3 = observable.subscribe(lambda x: print('O3: {0}'.format(x)))

loop = asyncio.get_event_loop()
try:
    print('STARTING LOOP')
    loop.run_forever()
except KeyboardInterrupt:
    pending_tasks = asyncio.Task.all_tasks()
    for task in pending_tasks:
        task.cancel()

    loop.run_until_complete(asyncio.gather(*pending_tasks))
    loop.close()

print('THE END')