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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

#![allow(clippy::disallowed_names)]
#![allow(clippy::let_underscore_future)]

use fred::prelude::*;
use std::time::Duration;
use tokio::time::sleep;

/// Examples showing how to set up keyspace notifications with clustered or centralized/sentinel deployments.
///
/// The most complicated part of this process involves safely handling reconnections. Keyspace events rely on the
/// pubsub interface, and clients are required to subscribe or resubscribe whenever a new connection is created. These
/// examples show how to manually handle reconnections, but the caller can also use the `SubscriberClient` interface
/// to remove some of the boilerplate.
///
/// If callers do not need the keyspace subscriptions to survive reconnects then the process is more
/// straightforward.
///
/// Both examples assume that the server has been configured to emit keyspace events (via `notify-keyspace-events`).
#[tokio::main]
async fn main() -> Result<(), Error> {
  clustered_keyspace_events().await?;
  centralized_keyspace_events().await?;
  Ok(())
}

async fn fake_traffic(client: &Client, amount: usize) -> Result<(), Error> {
  // use a new client since the provided client is subscribed to keyspace events
  let client = client.clone_new();
  client.init().await?;

  for idx in 0 .. amount {
    let key: Key = format!("foo-{}", idx).into();

    let _: () = client.set(&key, 1, None, None, false).await?;
    let _: () = client.incr(&key).await?;
    let _: () = client.del(&key).await?;
  }

  client.quit().await?;
  Ok(())
}

async fn centralized_keyspace_events() -> Result<(), Error> {
  let subscriber = Builder::default_centralized().build()?;

  let reconnect_subscriber = subscriber.clone();
  // resubscribe to the foo- prefix whenever we reconnect to a server
  let _reconnect_task = tokio::spawn(async move {
    let mut reconnect_rx = reconnect_subscriber.reconnect_rx();

    while let Ok(server) = reconnect_rx.recv().await {
      println!("Reconnected to {}. Subscribing to keyspace events...", server);
      reconnect_subscriber.psubscribe("__key__*:foo*").await?;
    }

    Ok::<_, Error>(())
  });

  // connect after setting up the reconnection logic
  subscriber.init().await?;

  let mut keyspace_rx = subscriber.keyspace_event_rx();
  // set up a task that listens for keyspace events
  let _keyspace_task = tokio::spawn(async move {
    while let Ok(event) = keyspace_rx.recv().await {
      println!(
        "Recv: {} on {} in db {}",
        event.operation,
        event.key.as_str_lossy(),
        event.db
      );
    }

    Ok::<_, Error>(())
  });

  // generate fake traffic and wait a second
  fake_traffic(&subscriber, 1_000).await?;
  sleep(Duration::from_secs(1)).await;
  subscriber.quit().await?;
  Ok(())
}

async fn clustered_keyspace_events() -> Result<(), Error> {
  let subscriber = Builder::default_clustered().build()?;

  let reconnect_subscriber = subscriber.clone();
  // resubscribe to the foo- prefix whenever we reconnect to a server
  let _reconnect_task = tokio::spawn(async move {
    let mut reconnect_rx = reconnect_subscriber.reconnect_rx();

    // in 7.x the reconnection interface added a `Server` struct to reconnect events to make this easier.
    while let Ok(server) = reconnect_rx.recv().await {
      println!("Reconnected to {}. Subscribing to keyspace events...", server);
      reconnect_subscriber
        .with_cluster_node(server)
        .psubscribe("__key__*:foo*")
        .await?;
    }

    Ok::<_, Error>(())
  });

  // connect after setting up the reconnection logic
  subscriber.init().await?;

  let mut keyspace_rx = subscriber.keyspace_event_rx();
  // set up a task that listens for keyspace events
  let _keyspace_task = tokio::spawn(async move {
    while let Ok(event) = keyspace_rx.recv().await {
      println!(
        "Recv: {} on {} in db {}",
        event.operation,
        event.key.as_str_lossy(),
        event.db
      );
    }

    Ok::<_, Error>(())
  });

  // generate fake traffic and wait a second
  fake_traffic(&subscriber, 1_000).await?;
  sleep(Duration::from_secs(1)).await;
  subscriber.quit().await?;
  Ok(())
}