|
| 1 | +use numaflow::sink::{self, Response, SinkRequest, Sinker}; |
| 2 | +use redis::AsyncCommands; |
| 3 | +use std::env; |
| 4 | +use tokio::sync::Mutex; |
| 5 | + |
| 6 | +/// RedisTestSink is a sink that writes messages to Redis hashes. |
| 7 | +/// Created for numaflow e2e tests. |
| 8 | +struct RedisTestSink { |
| 9 | + /// Redis hash key to store messages. This is set by an environment variable `SINK_HASH_KEY` |
| 10 | + /// under the sink container spec. |
| 11 | + hash_key: String, |
| 12 | + /// Used to determine how many subsequent number of messages at a time to check the order of. |
| 13 | + /// This is set by an environment variable `MESSAGE_COUNT` |
| 14 | + message_count: usize, |
| 15 | + /// Used to collect `message_count` number of messages, check whether they all arrived in order |
| 16 | + /// and increment the count for the order result in Redis |
| 17 | + inflight_messages: Mutex<Vec<SinkRequest>>, |
| 18 | + client: redis::Client, |
| 19 | + /// If true, checks the order of messages based on event time. |
| 20 | + /// This is set by an environment variable `CHECK_ORDER` |
| 21 | + check_order: bool, |
| 22 | +} |
| 23 | + |
| 24 | +impl RedisTestSink { |
| 25 | + /// Creates a new instance of RedisTestSink with a Redis client. |
| 26 | + fn new() -> Self { |
| 27 | + let client = |
| 28 | + redis::Client::open("redis://redis:6379").expect("Failed to create Redis client"); |
| 29 | + |
| 30 | + let hash_key = |
| 31 | + env::var("SINK_HASH_KEY").expect("SINK_HASH_KEY environment variable is not set"); |
| 32 | + |
| 33 | + let message_count: usize = env::var("MESSAGE_COUNT") |
| 34 | + .ok() |
| 35 | + .and_then(|s| s.parse().ok()) |
| 36 | + .unwrap_or(0); |
| 37 | + |
| 38 | + let check_order: bool = env::var("CHECK_ORDER") |
| 39 | + .ok() |
| 40 | + .and_then(|s| s.parse().ok()) |
| 41 | + .unwrap_or(false); |
| 42 | + |
| 43 | + RedisTestSink { |
| 44 | + client, |
| 45 | + hash_key, |
| 46 | + message_count, |
| 47 | + inflight_messages: Mutex::new(Vec::with_capacity(message_count)), |
| 48 | + check_order, |
| 49 | + } |
| 50 | + } |
| 51 | +} |
| 52 | + |
| 53 | +#[tonic::async_trait] |
| 54 | +impl Sinker for RedisTestSink { |
| 55 | + /// This redis UDSink is created for numaflow e2e tests. This handle function assumes that |
| 56 | + /// a redis instance listening on address redis:6379 has already been up and running. |
| 57 | + async fn sink(&self, mut input: tokio::sync::mpsc::Receiver<SinkRequest>) -> Vec<Response> { |
| 58 | + let mut results: Vec<Response> = Vec::new(); |
| 59 | + |
| 60 | + // Get async connection to Redis |
| 61 | + let mut con = self |
| 62 | + .client |
| 63 | + .get_multiplexed_async_connection() |
| 64 | + .await |
| 65 | + .expect("Failed to get Redis connection"); |
| 66 | + |
| 67 | + while let Some(datum) = input.recv().await { |
| 68 | + let id = datum.id.clone(); |
| 69 | + let value = datum.value.clone(); |
| 70 | + |
| 71 | + if self.check_order { |
| 72 | + let mut inflight = self.inflight_messages.lock().await; |
| 73 | + inflight.push(datum); |
| 74 | + |
| 75 | + if inflight.len() == self.message_count { |
| 76 | + // Check if messages are ordered by event time |
| 77 | + let ordered = inflight |
| 78 | + .windows(2) |
| 79 | + .all(|w| w[0].event_time <= w[1].event_time); |
| 80 | + |
| 81 | + let result_message = if ordered { "ordered" } else { "not ordered" }; |
| 82 | + |
| 83 | + // Increment the count for the order result in Redis |
| 84 | + let result: Result<(), redis::RedisError> = |
| 85 | + con.hincr(&self.hash_key, result_message, 1).await; |
| 86 | + |
| 87 | + match result { |
| 88 | + Ok(_) => { |
| 89 | + println!( |
| 90 | + "Incremented by 1 the no. of occurrences of {} under hash key {}", |
| 91 | + result_message, self.hash_key |
| 92 | + ); |
| 93 | + } |
| 94 | + Err(e) => { |
| 95 | + eprintln!("Set Error - {:?}", e); |
| 96 | + } |
| 97 | + } |
| 98 | + |
| 99 | + // Reset the inflight messages |
| 100 | + inflight.clear(); |
| 101 | + } |
| 102 | + } |
| 103 | + |
| 104 | + // Watermark and event time of the message can be accessed |
| 105 | + let _ = datum.event_time; |
| 106 | + let _ = datum.watermark; |
| 107 | + |
| 108 | + // We use redis hashes to store messages. |
| 109 | + // Each field of a hash is the content of a message and |
| 110 | + // value of the field is the no. of occurrences of the message. |
| 111 | + let value_str = String::from_utf8(value).unwrap_or_else(|_| "".to_string()); |
| 112 | + |
| 113 | + let result: Result<(), redis::RedisError> = |
| 114 | + con.hincr(&self.hash_key, &value_str, 1).await; |
| 115 | + |
| 116 | + match result { |
| 117 | + Ok(_) => { |
| 118 | + println!( |
| 119 | + "Incremented by 1 the no. of occurrences of {} under hash key {}", |
| 120 | + value_str, self.hash_key |
| 121 | + ); |
| 122 | + } |
| 123 | + Err(e) => { |
| 124 | + eprintln!("Set Error - {:?}", e); |
| 125 | + } |
| 126 | + } |
| 127 | + |
| 128 | + results.push(Response::ok(id)); |
| 129 | + } |
| 130 | + |
| 131 | + results |
| 132 | + } |
| 133 | +} |
| 134 | + |
| 135 | +#[tokio::main] |
| 136 | +async fn main() { |
| 137 | + let sink = RedisTestSink::new(); |
| 138 | + |
| 139 | + let server = sink::Server::new(sink); |
| 140 | + |
| 141 | + if let Err(e) = server.start().await { |
| 142 | + panic!("Failed to start sink function server: {:?}", e); |
| 143 | + } |
| 144 | +} |
0 commit comments