photon-backend 0.1.0

Photon publish/subscribe backend implementations and storage adapters
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

//! Multiplex assigned virtual shard streams for consumer-group subscribers.

use std::collections::HashMap;
use std::hash::BuildHasher;
use std::pin::Pin;
use std::sync::Arc;

use async_stream::stream;
use futures::stream::Stream;
use tokio::sync::mpsc;

use crate::backend::PhotonBackend;
use crate::error::Result;
use crate::models::Event;
use crate::shard_router::shard_storage_key;

/// Merge tail streams for `shard_ids`, each with its own replay cursor.
pub fn merge_shard_streams<S: BuildHasher + Send + Sync + 'static>(
    backend: Arc<dyn PhotonBackend>,
    topic_name: String,
    shard_ids: &[u32],
    after_seq_by_shard: HashMap<u32, Option<i64>, S>,
) -> Pin<Box<dyn Stream<Item = Result<Event>> + Send>> {
    let shards: Vec<u32> = shard_ids.to_vec();
    Box::pin(stream! {
        if shards.is_empty() {
            return;
        }
        let (tx, mut rx) = mpsc::channel::<Result<Event>>(256);
        for shard_id in shards {
            let key = shard_storage_key(shard_id);
            let after = after_seq_by_shard.get(&shard_id).copied().flatten();
            let topic = topic_name.clone();
            let backend = Arc::clone(&backend);
            let mut sub = backend.subscribe(topic, Some(key), after);
            let tx = tx.clone();
            tokio::spawn(async move {
                while let Some(item) = futures::StreamExt::next(&mut sub).await {
                    if tx.send(item).await.is_err() {
                        break;
                    }
                }
            });
        }
        drop(tx);
        while let Some(item) = rx.recv().await {
            yield item;
        }
    })
}