Communicating between 2 observers

[yoavweber]

Hey people,
How do I communicate with 2 observers?

I have 2 observers that are listing to a WebSocket stream that emits different types of events: new_order event and leverage_change event.

Both events execute an async function.

If I am receiving a leverage_change I must wait until the execution of the async function is finished before I could run the new_order event. So I need to find a way to notify the new order observer when he can start/stop working.

Does anyone have any design suggestions on how to achieve such a goal?
I drafted a simple code snippet to show the problem:

# --- creating observer for each event ---

leverage_change_obs = websocket_obs.pipe(
    ops.filter(lambda data: data == "leverage_change"))


new_order_obs = websocket_obs.pipe(
    ops.filter(lambda data: data == "new_order"))


async def leverage_change(data):
    print("started leverage change")
    await asyncio.sleep(3)
    print("finished leverage change")


async def new_order(data):
    print("started new order")
    await asyncio.sleep(1)
    print("finished new order")


def run():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    leverage_change_obs.subscribe(
        on_next=lambda i: leverage_change(i, loop),
        on_error=lambda e: print("Error : {0}".format(e)),
        on_completed=lambda: print("Job Done!"),
    )
	
    new_order_obs.subscribe(
        on_next=lambda i: new_order(i, loop),
        on_error=lambda e: print("Error : {0}".format(e)),
        on_completed=lambda: print("Job Done!"),
    )
    loop.run_forever()


run()

And this is the full code if anyone would like to run it:

import asyncio
import reactivex as rx
from reactivex import operators as ops

# mocking the websocket events
event_list = ["new_order", "leverage_change", ,"other_event","new_order",
              "new_order", "leverage_change", "other_event","new_order"]

websocket_obs = rx.from_list(event_list)


leverage_change_obs = websocket_obs.pipe(
    ops.filter(lambda data: data == "leverage_change"))


new_order_obs = websocket_obs.pipe(
    ops.filter(lambda data: data == "new_order"))


def leverage_wrapper(data, loop):
    loop.create_task(leverage_change(data))


def new_order_wrapper(data, loop):
    loop.create_task(new_order(data))

async def leverage_change(data):
    print("started leverage change")
    await asyncio.sleep(3)
    print("finished leverage change")


async def new_order(data):
    print("started new order")
    await asyncio.sleep(1)
    print("finished new order")


def run():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
	#notify when to start worknig
    leverage_change_obs.subscribe(
        on_next=lambda i: leverage_wrapper(i, loop),
        on_error=lambda e: print("Error : {0}".format(e)),
        on_completed=lambda: print("Job Done!"),
    )
	
    new_order_obs.subscribe(
        on_next=lambda i: new_order_wrapper(i, loop),
        on_error=lambda e: print("Error : {0}".format(e)),
        on_completed=lambda: print("Job Done!"),
    )
    loop.run_forever()


run()