wingfoil 8.0.0

graph based stream processing framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

//! Redis Pub/Sub producer — subscribes to a channel and streams messages.

use super::{RedisConnection, RedisEvent};
use crate::nodes::{RunParams, produce_async};
use crate::types::*;
use futures::StreamExt;
use std::rc::Rc;

/// Subscribe to a Redis `channel` and stream every message published to it as a [`RedisEvent`].
///
/// Connects and issues `SUBSCRIBE` once at startup, then emits each incoming message.
/// Redis Pub/Sub has no backlog: only messages published *after* the subscription is
/// registered are delivered. Any connection or subscribe error terminates the producer
/// and propagates to the graph.
///
/// Emits `Burst<RedisEvent>`. Use `.collapse()` for single-event processing.
///
/// Designed for `RunMode::RealTime`. In `HistoricalFrom` mode timestamps are wall-clock
/// `NanoTime::now()` rather than historical, since messages arrive live off the wire.
#[must_use]
pub fn redis_sub(
    connection: impl Into<RedisConnection>,
    channel: impl Into<String>,
) -> Rc<dyn Stream<Burst<RedisEvent>>> {
    let connection = connection.into();
    let channel = channel.into();
    produce_async(
        move |_ctx: RunParams| async move {
            let client = redis::Client::open(connection.url.as_str())
                .map_err(|e| anyhow::anyhow!("redis client open failed: {e}"))?;
            let mut pubsub = client
                .get_async_pubsub()
                .await
                .map_err(|e| anyhow::anyhow!("redis connect failed: {e}"))?;
            pubsub
                .subscribe(channel.as_str())
                .await
                .map_err(|e| anyhow::anyhow!("redis subscribe to {channel:?} failed: {e}"))?;

            Ok(async_stream::stream! {
                let mut messages = pubsub.on_message();
                while let Some(msg) = messages.next().await {
                    let event = RedisEvent {
                        channel: msg.get_channel_name().to_string(),
                        payload: msg.get_payload_bytes().to_vec(),
                    };
                    yield Ok((NanoTime::now(), event));
                }
                // `on_message` ends only when the connection drops.
                yield Err(anyhow::anyhow!("redis subscription stream closed unexpectedly"));
            })
        },
        None,
    )
}