Provide example with global stream for subscription about async-graphql

[sunli829]

@atik7 https://github.com/async-graphql/examples/tree/c8219078a4b7aa6d84d22e9b79f033088897be4b/poem/subscription-redis

[sunli829]

The problem has been solved.😁

[sunli829]

Okay, I will provide it later today. @atik7

[sunli829]

Global stream are easy to implement, and I'll add a component tomorrow to do just that.

Creating the stream should be an asynchronous function, so I'm going to do that first.

[sunli829]

Are you Chinese?

[nicolaiunrein]

Are you Chinese?

No German 😊
Why?

[sunli829]

I thought we were in the same time zone.😄

I added SimpeBroker and updated the example.
SimpleBroker should be used for standalone testing, and later I will provide support for Kafka, Redis, etc.

[nicolaiunrein]

Yeah thought so too. 😆 Are you from china?

[sunli829]

Yes, I'm Chinese.😁

[nicolaiunrein]

SimpleBroker panics at called `Option::unwrap()` on a `None` value if used with more than one type. Is this intended? If so this should be reflected in the docs/example.

[sunli829]

I didn't have a test because I was going to sleep. That should be fine.

[maz000]

Can you provide a “Redis-based” (https://github.com/mitsuhiko/redis-rs) example of global stream for subscription?

[lostb1t]

I would be interested in an redis-rs example as well.

[bponidev]

May be helpful...

publishing

let message = serde_json::to_string(&order)?;
let channel = format!("user_{}_order", driver.user_id);
let _ = caching::cache_publish(ctx, channel, message).await?;

and consuming

async fn logistics_order(
    &self,
    ctx: &Context<'_>,
    channel: String,
) -> impl Stream<Item = Order> {
    let context = ctx.data::<MainContext>().unwrap();
    let con = context.redis.get_async_connection().await.unwrap();
    let mut pubsub = con.into_pubsub();
    pubsub.subscribe(channel).await.unwrap();
    stream! {
    let mut msg_stream = pubsub.on_message();
        while let Some(msg) = msg_stream.next().await {
            let message: String = msg.get_payload().unwrap();
            let order: Order = serde_json::from_str(&message).unwrap();
            yield order;
        }
    }
}

[lostb1t]

Appreciated, actually just finished an similar implementation but using rust channels.

Because the above example creates an redis connection for each subscription. (You cannot share a pubsub channel accross threads)

So what i did was to fanout pubsub events to subscriptions with rust channels

[bponidev]

Can u provide some code example, it would be greatly helpful for us.

[lostb1t]

...

lazy_static! {
  pub static ref EVENT_BUS: tokio::sync::broadcast::Sender<Event> = {
    let (tx, _) = tokio::sync::broadcast::channel(10000);
    tx
  };
}

...

#[Object]
impl Mutation {
  async fn emit_event(&self, ctx: &Context<'_>, input: Input) -> Result<Output> {
    let mut redis_conn = ctx
      .data_unchecked::<redis::aio::MultiplexedConnection>()
      .clone();
    let _: () = redis_conn
      .publish("events", serde_json::to_string(&input)?)
      .await?;

    Ok(input)
  }
}

...

#[Subscription]
impl Subscription {
  async fn receive_event(&self, ctx: &Context<'_>) -> impl Stream<Item = Event> {
    BroadcastStream::new(EVENT_BUS.subscribe())
      .filter(|msg| msg.is_ok())
      .map(|msg| msg.unwrap())
  }
}

....


#[tokio::main]
async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
  let redis_url = env::var("REDIS_URL").expect("REDIS_URL is not set");
  let redis_client = redis::Client::open(redis_url).expect("client");
  let mut redis_pubsub = redis_client
    .clone()
    .get_tokio_connection()
    .await?
    .into_pubsub();
  let fanout_task = async move {
    redis_pubsub
      .subscribe("events")
      .await
      .unwrap();
    let mut msg_stream = redis_pubsub.into_on_message();
    while let Some(msg) = msg_stream.next().await {
      let msg: String = msg.get_payload().unwrap();
      let event: Event = serde_json::from_str(&msg).unwrap();
      let _ = schema::EVENT_BUS.send(event);
    }
  };
  // spawn the fanout task as a background task that will run forever.
  tokio::spawn(fanout_task);

  ...

}

[atik7]

@sunli829 Please add a redis example, thank you!

[atik7]

@sunli829 thank you :-)