v4 refactor help 🙏

[tomjohnhall]

I'm attempting to update DjangoCHannelsGraphqlWs to work with updated versions of graphql + graphene. In the process I'm bumping rx from 1.6.1 to reactivex 4.

Would be grateful if anyone could explain to me what is happening in the following code snippet, and / or how the equivalent can be done in version 4 -

trigger = rx.subjects.Subject()

stream = trigger.map(publish_callback).filter(  # pylint: disable=no-member
            lambda publish_returned: publish_returned is not self.SKIP
        )
        
 # this next part is run elsewhere as part of a continuous loop on a worker thread 
 # notification queue is an asyncio.Queue(), presumably receiving messages to be passed to listening websocket clients
 payload = await notification_queue.get()
 trigger.on_next(Serializer.deserialize(payload))

I think I have a general grasp of what's going on but any help translating would be much appreciated!

[MainRo]

since rxpy v3, the way to use operators has changed. You need something like this:

import reactivex as rx
from reactivex import operators as ops

trigger = rx.subjects.Subject()

stream = trigger.pipe(
    ops.map(publish_callback),
    ops.filter(  # pylint: disable=no-member
            lambda publish_returned: publish_returned is not self.SKIP
    ),
)

[tomjohnhall]

Perfect, thank you for this - I saw this flow in the docs and had a hunch it might be the replacement but without a clear understanding of what was going on in the background I couldn't be sure.

Much appreciated! 🙌