fred 10.1.0

An async client for Redis and Valkey.
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

use crate::error::Error;
use glommio::{
  channels::local_channel::{new_unbounded, LocalReceiver, LocalSender},
  GlommioError,
  ResourceType,
};
use std::{cell::RefCell, collections::BTreeMap, rc::Rc};

struct Inner<T: Clone> {
  pub counter: u64,
  pub senders: BTreeMap<u64, LocalSender<T>>,
}

/// A multi-producer multi-consumer channel receiver.
///
/// See [LocalReceiver](glommio::channels::local_channel::LocalReceiver) for more information.
pub struct BroadcastReceiver<T: Clone> {
  id:    u64,
  inner: Rc<RefCell<Inner<T>>>,
  rx:    LocalReceiver<T>,
}

impl<T: Clone> BroadcastReceiver<T> {
  /// Receives data from this channel.
  ///
  /// See [recv](glommio::channels::local_channel::LocalReceiver::recv) for more information.
  pub async fn recv(&self) -> Result<T, Error> {
    match self.rx.recv().await {
      Some(v) => Ok(v),
      None => Err(Error::new_canceled()),
    }
  }
}

impl<T: Clone> Drop for BroadcastReceiver<T> {
  fn drop(&mut self) {
    self.inner.as_ref().borrow_mut().senders.remove(&self.id);
  }
}

#[derive(Clone)]
pub struct BroadcastSender<T: Clone> {
  inner: Rc<RefCell<Inner<T>>>,
}

impl<T: Clone> BroadcastSender<T> {
  pub fn new() -> Self {
    BroadcastSender {
      inner: Rc::new(RefCell::new(Inner {
        counter: 0,
        senders: BTreeMap::new(),
      })),
    }
  }

  pub fn subscribe(&self) -> BroadcastReceiver<T> {
    let (tx, rx) = new_unbounded();
    let id = {
      let mut guard = self.inner.as_ref().borrow_mut();
      let count = guard.counter.wrapping_add(1);
      guard.counter = count;
      guard.senders.insert(count, tx);
      guard.counter
    };

    BroadcastReceiver {
      id,
      rx,
      inner: self.inner.clone(),
    }
  }

  pub fn send<F: Fn(&T)>(&self, msg: &T, func: F) {
    let mut guard = self.inner.as_ref().borrow_mut();

    let to_remove: Vec<u64> = guard
      .senders
      .iter()
      .filter_map(|(id, tx)| {
        if let Err(GlommioError::Closed(ResourceType::Channel(val))) = tx.try_send(msg.clone()) {
          func(&val);
          Some(*id)
        } else {
          None
        }
      })
      .collect();

    for id in to_remove.into_iter() {
      guard.senders.remove(&id);
    }
  }
}