Skip to main content

nominal_streaming/
lib.rs

1/*!
2`nominal-streaming` is a crate for streaming data into [Nominal Core](https://nominal.io/products/core).
3
4The library aims to balance three concerns:
5
61. Data should exist in-memory only for a limited, configurable amount of time before it's sent to Core.
71. Writes should fall back to disk if there are network failures.
81. Backpressure should be applied to incoming requests when network throughput is saturated.
9
10This library streams data to Nominal Core, to a file, or to Nominal Core with a file as backup (recommended to protect against network failures).
11It also provides configuration to manage the tradeoff between above listed concerns.
12
13<div class="warning">
14This library is still under active development and may make breaking changes.
15</div>
16
17## Conceptual overview
18
19Data is sent to a [Stream](https://docs.rs/nominal-streaming/latest/nominal_streaming/stream/struct.NominalDatasetStream.html) via a Writer.
20For example:
21
22- A file stream is constructed as:
23
24  ```rust,no_run
25  use nominal_streaming::stream::NominalDatasetStreamBuilder;
26
27  let stream = NominalDatasetStreamBuilder::new()
28      .stream_to_file("my_data.avro")
29      .build();
30  ```
31
32- A stream that sends data to Nominal Core, but writes failed requests to a file, is created as follows:
33
34  ```rust,ignore
35  let stream = NominalDatasetStreamBuilder::new()
36      .stream_to_core(token, dataset_rid, handle)
37      .with_file_fallback("fallback.avro")
38      .build();
39  ```
40
41- Or, you can build a stream that sends data to Nominal Core *and* to a file:
42
43  ```rust,ignore
44  let stream = NominalDatasetStreamBuilder::new()
45      .stream_to_core(token, dataset_rid, handle)
46      .stream_to_file("my_data.avro")
47      .build();
48  ```
49
50(See below for a [full example](#example-streaming-from-memory-to-nominal-core-with-file-fallback), that also shows how to create the `token`, `dataset_rid`, and `handle` values above.)
51
52Once we have a Stream, we can construct a Writer and send values to it:
53
54```rust,ignore
55let channel_descriptor = ChannelDescriptor::with_tags(
56    "channel_1", [("experiment_id", "123")]
57);
58
59let mut writer = stream.double_writer(channel_descriptor);
60
61// Stream single data point
62let start_time = UNIX_EPOCH.elapsed().unwrap();
63let value: f64 = 123;
64writer.push(start_time, value);
65```
66
67Here, we are enqueuing data onto Channel 1, with tags "name" and "batch".
68These are, of course, just examples, and you can choose your own.
69
70## Example: streaming from memory to Nominal Core, with file fallback
71
72This is the typical scenario where we want to stream some values from memory into a [Nominal Dataset](https://docs.nominal.io/core/sdk/python-client/streaming/overview#streaming-data-to-a-dataset).
73If the upload fails (say because of network errors), we'd like to instead send the data to an Avro file. Note that the Avro spec does not support uint64 values, so those will be stored as signed int64 values.
74
75Note that we set up the async [Tokio runtime](https://tokio.rs/), since that is required by the underlying [`NominalCoreConsumer`](https://docs.rs/nominal-streaming/latest/nominal_streaming/consumer/struct.NominalCoreConsumer.html).
76
77```rust,no_run
78use nominal_streaming::prelude::*;
79use nominal_streaming::stream::NominalDatasetStreamBuilder;
80
81use std::time::UNIX_EPOCH;
82
83
84static DATASET_RID: &str = "ri.catalog....";  // your dataset ID here
85
86
87fn main() {
88    // The NominalCoreConsumer requires a tokio runtime
89    tokio::runtime::Builder::new_multi_thread()
90        .enable_all()
91        .worker_threads(4)
92        .thread_name("tokio")
93        .build()
94        .expect("Failed to create Tokio runtime")
95        .block_on(async_main());
96}
97
98
99async fn async_main() {
100    // Configure token for authentication
101    let token = BearerToken::new(
102        std::env::var("NOMINAL_TOKEN")
103            .expect("NOMINAL_TOKEN environment variable not set")
104            .as_str(),
105    )
106    .expect("Invalid token");
107
108    let dataset_rid = ResourceIdentifier::new(DATASET_RID).unwrap();
109    let handle = tokio::runtime::Handle::current();
110
111    let stream = NominalDatasetStreamBuilder::new()
112        .stream_to_core(token, dataset_rid, handle)
113        .with_file_fallback("fallback.avro")
114        .build();
115
116    let channel_descriptor = ChannelDescriptor::with_tags("channel_1", [("experiment_id", "123")]);
117
118    let mut writer = stream.double_writer(channel_descriptor);
119
120    // Generate and upload 100,000 data points
121    for i in 0..100_000 {
122        let start_time = UNIX_EPOCH.elapsed().unwrap();
123        let value = i % 50;
124        writer.push(start_time, value as f64);
125    }
126}
127```
128
129## Additional configuration
130
131### Stream options
132
133Above, you saw an example using [`NominalStreamOpts::default`](https://docs.rs/nominal-streaming/latest/nominal_streaming/stream/struct.NominalStreamOpts.html).
134The following stream options can be set using `.with_options(...)` on the StreamBuilder:
135
136```text
137NominalStreamOpts {
138  max_points_per_record: usize,
139  max_request_delay: Duration,
140  max_buffered_requests: usize,
141  request_dispatcher_tasks: usize,
142}
143```
144
145### Logging errors
146
147Most of the time, when things go wrong, we want some form of reporting. You can enable debug logging on the StreamBuilder by using `.enable_logging()`:
148
149```rust,ignore
150let stream = NominalDatasetStreamBuilder::new()
151    .stream_to_core(token, dataset_rid, handle)
152    .with_file_fallback("fallback.avro")
153    .enable_logging()
154    .build();
155```
156*/
157
158pub mod client;
159pub mod consumer;
160pub mod listener;
161#[cfg(test)]
162mod simulated_consumer;
163pub mod stream;
164pub mod types;
165pub mod upload;
166
167pub use nominal_api as api;
168
169/// This includes the most common types in this crate, re-exported for your convenience.
170pub mod prelude {
171    pub use conjure_object::BearerToken;
172    pub use conjure_object::ResourceIdentifier;
173    pub use nominal_api::tonic::google::protobuf::Timestamp;
174    pub use nominal_api::tonic::io::nominal::scout::api::proto::array_points::ArrayType;
175    pub use nominal_api::tonic::io::nominal::scout::api::proto::points::PointsType;
176    pub use nominal_api::tonic::io::nominal::scout::api::proto::ArrayPoints;
177    pub use nominal_api::tonic::io::nominal::scout::api::proto::DoubleArrayPoint;
178    pub use nominal_api::tonic::io::nominal::scout::api::proto::DoubleArrayPoints;
179    pub use nominal_api::tonic::io::nominal::scout::api::proto::DoublePoint;
180    pub use nominal_api::tonic::io::nominal::scout::api::proto::DoublePoints;
181    pub use nominal_api::tonic::io::nominal::scout::api::proto::IntegerPoint;
182    pub use nominal_api::tonic::io::nominal::scout::api::proto::IntegerPoints;
183    pub use nominal_api::tonic::io::nominal::scout::api::proto::StringArrayPoint;
184    pub use nominal_api::tonic::io::nominal::scout::api::proto::StringArrayPoints;
185    pub use nominal_api::tonic::io::nominal::scout::api::proto::StringPoint;
186    pub use nominal_api::tonic::io::nominal::scout::api::proto::StringPoints;
187    pub use nominal_api::tonic::io::nominal::scout::api::proto::StructPoint;
188    pub use nominal_api::tonic::io::nominal::scout::api::proto::StructPoints;
189    pub use nominal_api::tonic::io::nominal::scout::api::proto::Uint64Point;
190    pub use nominal_api::tonic::io::nominal::scout::api::proto::Uint64Points;
191    pub use nominal_api::tonic::io::nominal::scout::api::proto::WriteRequest;
192    pub use nominal_api::tonic::io::nominal::scout::api::proto::WriteRequestNominal;
193
194    pub use crate::consumer::NominalCoreConsumer;
195    pub use crate::stream::NominalDatasetStream;
196    #[expect(deprecated)]
197    pub use crate::stream::NominalDatasourceStream;
198    pub use crate::stream::NominalStreamOpts;
199    pub use crate::types::AuthProvider;
200    pub use crate::types::ChannelDescriptor;
201    pub use crate::types::IntoTimestamp;
202}
203
204#[cfg(test)]
205mod tests {
206    use std::collections::HashMap;
207    use std::sync::Arc;
208    use std::sync::Mutex;
209    use std::thread;
210    use std::time::Duration;
211    use std::time::UNIX_EPOCH;
212
213    use nominal_api::tonic::io::nominal::scout::api::proto::array_points::ArrayType;
214    use nominal_api::tonic::io::nominal::scout::api::proto::ArrayPoints;
215    use nominal_api::tonic::io::nominal::scout::api::proto::IntegerPoint;
216    use nominal_api::tonic::io::nominal::scout::api::proto::Points;
217
218    use crate::client::PRODUCTION_API_URL;
219    use crate::consumer::ConsumerResult;
220    use crate::consumer::RequestConsumerWithFallback;
221    use crate::consumer::WriteRequestConsumer;
222    use crate::prelude::*;
223    use crate::simulated_consumer::SimulatedNetworkConfig;
224    use crate::simulated_consumer::SimulatedNetworkConsumer;
225    use crate::simulated_consumer::SimulatedNetworkFailure;
226    use crate::simulated_consumer::SimulatedRetryPolicy;
227
228    #[derive(Debug)]
229    struct TestDatasourceStream {
230        requests: Mutex<Vec<WriteRequestNominal>>,
231    }
232
233    impl WriteRequestConsumer for Arc<TestDatasourceStream> {
234        fn consume(&self, request: &WriteRequestNominal) -> ConsumerResult<()> {
235            self.requests.lock().unwrap().push(request.clone());
236            Ok(())
237        }
238    }
239
240    fn create_test_stream() -> (Arc<TestDatasourceStream>, NominalDatasetStream) {
241        let test_consumer = Arc::new(TestDatasourceStream {
242            requests: Mutex::new(vec![]),
243        });
244        let stream = create_stream_with_consumer(test_consumer.clone(), 1000);
245
246        (test_consumer, stream)
247    }
248
249    fn create_stream_with_consumer<C>(
250        consumer: C,
251        max_points_per_record: usize,
252    ) -> NominalDatasetStream
253    where
254        C: WriteRequestConsumer + 'static,
255    {
256        let stream = NominalDatasetStream::new_with_consumer(
257            consumer,
258            NominalStreamOpts {
259                max_points_per_record,
260                max_request_delay: Duration::from_millis(100),
261                max_buffered_requests: 2,
262                request_dispatcher_tasks: 4,
263                base_api_url: PRODUCTION_API_URL.to_string(),
264            },
265        );
266
267        stream
268    }
269
270    fn create_stream_with_consumer_and_options<C>(
271        consumer: C,
272        opts: NominalStreamOpts,
273    ) -> NominalDatasetStream
274    where
275        C: WriteRequestConsumer + 'static,
276    {
277        NominalDatasetStream::new_with_consumer(consumer, opts)
278    }
279
280    fn total_double_points(requests: &[WriteRequestNominal], channel_name: &str) -> usize {
281        requests
282            .iter()
283            .flat_map(|request| request.series.iter())
284            .filter(|series| {
285                series
286                    .channel
287                    .as_ref()
288                    .is_some_and(|channel| channel.name == channel_name)
289            })
290            .map(|series| {
291                let Some(Points {
292                    points_type: Some(PointsType::DoublePoints(points)),
293                }) = series.points.as_ref()
294                else {
295                    return 0;
296                };
297                points.points.len()
298            })
299            .sum()
300    }
301
302    fn point_counts_by_channel(
303        requests: &[WriteRequestNominal],
304    ) -> HashMap<String, (&'static str, usize)> {
305        let mut counts = HashMap::new();
306
307        for series in requests.iter().flat_map(|request| request.series.iter()) {
308            let channel_name = series
309                .channel
310                .as_ref()
311                .expect("series should have a channel")
312                .name
313                .clone();
314            let points_type = series
315                .points
316                .as_ref()
317                .and_then(|points| points.points_type.as_ref())
318                .expect("series should have points");
319
320            let (kind, point_count) = match points_type {
321                PointsType::DoublePoints(points) => ("double", points.points.len()),
322                PointsType::StringPoints(points) => ("string", points.points.len()),
323                PointsType::IntegerPoints(points) => ("int", points.points.len()),
324                PointsType::Uint64Points(points) => ("uint64", points.points.len()),
325                PointsType::StructPoints(points) => ("struct", points.points.len()),
326                PointsType::ArrayPoints(ArrayPoints {
327                    array_type: Some(ArrayType::DoubleArrayPoints(points)),
328                }) => ("double_array", points.points.len()),
329                PointsType::ArrayPoints(ArrayPoints {
330                    array_type: Some(ArrayType::StringArrayPoints(points)),
331                }) => ("string_array", points.points.len()),
332                PointsType::ArrayPoints(ArrayPoints { array_type: None }) => ("array", 0),
333            };
334
335            let entry = counts.entry(channel_name).or_insert((kind, 0));
336            assert_eq!(entry.0, kind, "mismatched point type for channel");
337            entry.1 += point_count;
338        }
339
340        counts
341    }
342
343    #[test_log::test]
344    fn test_stream() {
345        let (test_consumer, stream) = create_test_stream();
346
347        for batch in 0..5 {
348            let mut points = Vec::new();
349            for i in 0..1000 {
350                let start_time = UNIX_EPOCH.elapsed().unwrap();
351                points.push(DoublePoint {
352                    timestamp: Some(Timestamp {
353                        seconds: start_time.as_secs() as i64,
354                        nanos: start_time.subsec_nanos() as i32 + i,
355                    }),
356                    value: (i % 50) as f64,
357                });
358            }
359
360            stream.enqueue(
361                &ChannelDescriptor::with_tags("channel_1", [("batch_id", batch.to_string())]),
362                points,
363            );
364        }
365
366        drop(stream); // wait for points to flush
367
368        let requests = test_consumer.requests.lock().unwrap();
369
370        // validate that the requests were flushed based on the max_records value, not the
371        // max request delay
372        assert_eq!(requests.len(), 5);
373        let series = requests.first().unwrap().series.first().unwrap();
374        if let Some(PointsType::DoublePoints(points)) =
375            series.points.as_ref().unwrap().points_type.as_ref()
376        {
377            assert_eq!(points.points.len(), 1000);
378        } else {
379            panic!("unexpected data type");
380        }
381    }
382
383    #[test_log::test]
384    fn test_stream_types() {
385        let (test_consumer, stream) = create_test_stream();
386
387        for batch in 0..5 {
388            let mut doubles = Vec::new();
389            let mut strings = Vec::new();
390            let mut structs = Vec::new();
391            let mut ints = Vec::new();
392            let mut uints = Vec::new();
393            let mut double_arrays = Vec::new();
394            let mut string_arrays = Vec::new();
395            for i in 0..1000 {
396                let start_time = UNIX_EPOCH.elapsed().unwrap();
397                doubles.push(DoublePoint {
398                    timestamp: Some(start_time.into_timestamp()),
399                    value: (i % 50) as f64,
400                });
401                strings.push(StringPoint {
402                    timestamp: Some(start_time.into_timestamp()),
403                    value: format!("{}", i % 50),
404                });
405                structs.push(StructPoint {
406                    timestamp: Some(start_time.into_timestamp()),
407                    json_string: format!("{{\"v\":{}}}", i % 50),
408                });
409                ints.push(IntegerPoint {
410                    timestamp: Some(start_time.into_timestamp()),
411                    value: i % 50,
412                });
413                uints.push(Uint64Point {
414                    timestamp: Some(start_time.into_timestamp()),
415                    value: (i % 50) as u64,
416                });
417                double_arrays.push(DoubleArrayPoint {
418                    timestamp: Some(start_time.into_timestamp()),
419                    value: vec![(i % 50) as f64, ((i + 1) % 50) as f64],
420                });
421                string_arrays.push(StringArrayPoint {
422                    timestamp: Some(start_time.into_timestamp()),
423                    value: vec![format!("{}", i % 50)],
424                });
425            }
426
427            stream.enqueue(
428                &ChannelDescriptor::with_tags("double", [("batch_id", batch.to_string())]),
429                doubles,
430            );
431            stream.enqueue(
432                &ChannelDescriptor::with_tags("string", [("batch_id", batch.to_string())]),
433                strings,
434            );
435            stream.enqueue(
436                &ChannelDescriptor::with_tags("struct", [("batch_id", batch.to_string())]),
437                structs,
438            );
439            stream.enqueue(
440                &ChannelDescriptor::with_tags("int", [("batch_id", batch.to_string())]),
441                ints,
442            );
443            stream.enqueue(
444                &ChannelDescriptor::with_tags("uint64", [("batch_id", batch.to_string())]),
445                uints,
446            );
447            stream.enqueue(
448                &ChannelDescriptor::with_tags("double_array", [("batch_id", batch.to_string())]),
449                double_arrays,
450            );
451            stream.enqueue(
452                &ChannelDescriptor::with_tags("string_array", [("batch_id", batch.to_string())]),
453                string_arrays,
454            );
455        }
456
457        drop(stream); // wait for points to flush
458
459        let requests = test_consumer.requests.lock().unwrap();
460
461        // validate that the requests were flushed based on the max_records value, not the
462        // max request delay
463        assert_eq!(requests.len(), 35);
464
465        let r = requests
466            .iter()
467            .flat_map(|r| r.series.clone())
468            .map(|s| {
469                (
470                    s.channel.unwrap().name,
471                    s.points.unwrap().points_type.unwrap(),
472                )
473            })
474            .collect::<HashMap<_, _>>();
475        let PointsType::DoublePoints(dp) = r.get("double").unwrap() else {
476            panic!("invalid double points type");
477        };
478
479        let PointsType::IntegerPoints(ip) = r.get("int").unwrap() else {
480            panic!("invalid int points type");
481        };
482
483        let PointsType::Uint64Points(up) = r.get("uint64").unwrap() else {
484            panic!("invalid uint64 points type");
485        };
486
487        let PointsType::StringPoints(sp) = r.get("string").unwrap() else {
488            panic!("invalid string points type");
489        };
490
491        let PointsType::StructPoints(stp) = r.get("struct").unwrap() else {
492            panic!("invalid struct points type");
493        };
494
495        let PointsType::ArrayPoints(ArrayPoints {
496            array_type: Some(ArrayType::DoubleArrayPoints(dap)),
497        }) = r.get("double_array").unwrap()
498        else {
499            panic!("invalid double array points type");
500        };
501
502        let PointsType::ArrayPoints(ArrayPoints {
503            array_type: Some(ArrayType::StringArrayPoints(sap)),
504        }) = r.get("string_array").unwrap()
505        else {
506            panic!("invalid string array points type");
507        };
508
509        // collect() overwrites into a single request per channel
510        assert_eq!(dp.points.len(), 1000);
511        assert_eq!(sp.points.len(), 1000);
512        assert_eq!(ip.points.len(), 1000);
513        assert_eq!(up.points.len(), 1000);
514        assert_eq!(stp.points.len(), 1000);
515        assert_eq!(dap.points.len(), 1000);
516        assert_eq!(sap.points.len(), 1000);
517    }
518
519    #[test_log::test]
520    #[should_panic(expected = "mismatched types")]
521    fn test_mismatched_array_types_panics() {
522        // Protects the exhaustive match in SeriesBufferGuard::extend from being
523        // silently simplified to a catch-all: pushing a DoubleArray and then a
524        // StringArray to the same channel must panic at buffer merge time.
525        //
526        // ManuallyDrop skips NominalDatasetStream::drop during unwind — the panic
527        // leaves unflushed_points non-zero, which would cause drop to spin forever.
528        let (_test_consumer, stream) = create_test_stream();
529        let stream = std::mem::ManuallyDrop::new(stream);
530        let cd = ChannelDescriptor::new("mixed_array");
531
532        let ts = UNIX_EPOCH.elapsed().unwrap().into_timestamp();
533
534        stream.enqueue(
535            &cd,
536            vec![DoubleArrayPoint {
537                timestamp: Some(ts),
538                value: vec![1.0, 2.0],
539            }],
540        );
541        stream.enqueue(
542            &cd,
543            vec![StringArrayPoint {
544                timestamp: Some(ts),
545                value: vec!["a".into()],
546            }],
547        );
548    }
549
550    #[test_log::test]
551    fn test_writer() {
552        let (test_consumer, stream) = create_test_stream();
553
554        let cd = ChannelDescriptor::new("channel_1");
555        let mut writer = stream.double_writer(cd);
556
557        for i in 0..5000 {
558            let start_time = UNIX_EPOCH.elapsed().unwrap();
559            let value = i % 50;
560            writer.push(start_time, value as f64);
561        }
562
563        drop(writer); // flush points to stream
564        drop(stream); // flush stream to nominal
565
566        let requests = test_consumer.requests.lock().unwrap();
567
568        assert_eq!(requests.len(), 5);
569        let series = requests.first().unwrap().series.first().unwrap();
570        if let Some(PointsType::DoublePoints(points)) =
571            series.points.as_ref().unwrap().points_type.as_ref()
572        {
573            assert_eq!(points.points.len(), 1000);
574        } else {
575            panic!("unexpected data type");
576        }
577    }
578
579    #[test_log::test]
580    fn test_time_flush() {
581        let (test_consumer, stream) = create_test_stream();
582
583        let cd = ChannelDescriptor::new("channel_1");
584        let mut writer = stream.double_writer(cd);
585
586        writer.push(UNIX_EPOCH.elapsed().unwrap(), 1.0);
587        thread::sleep(Duration::from_millis(101));
588        writer.push(UNIX_EPOCH.elapsed().unwrap(), 2.0); // first flush
589        thread::sleep(Duration::from_millis(101));
590        writer.push(UNIX_EPOCH.elapsed().unwrap(), 3.0); // second flush
591
592        drop(writer);
593        drop(stream);
594
595        let requests = test_consumer.requests.lock().unwrap();
596        dbg!(&requests);
597        assert_eq!(requests.len(), 2);
598    }
599
600    #[test_log::test]
601    fn simulated_network_retries_transient_failures_and_delivers_data() {
602        let test_consumer = Arc::new(TestDatasourceStream {
603            requests: Mutex::new(vec![]),
604        });
605        let simulated_network = SimulatedNetworkConsumer::new(
606            test_consumer.clone(),
607            SimulatedNetworkConfig::default()
608                .with_latency(Duration::from_micros(10), Duration::from_micros(10))
609                .with_bandwidth_limit(32 * 1024 * 1024)
610                .with_failure_pattern(SimulatedNetworkFailure::FailFirstAttemptsPerRequest {
611                    attempts: 1,
612                })
613                .with_failure_pattern(SimulatedNetworkFailure::InitialOutageAttempts {
614                    attempts: 1,
615                })
616                .with_retry_policy(
617                    SimulatedRetryPolicy::new(2, Duration::from_micros(10))
618                        .with_jitter(Duration::from_micros(10)),
619                ),
620        );
621        let stats = simulated_network.stats();
622        let stream = create_stream_with_consumer(simulated_network, 4);
623
624        let cd = ChannelDescriptor::new("networked_double");
625        let mut writer = stream.double_writer(cd);
626        for i in 0..12 {
627            writer.push(UNIX_EPOCH.elapsed().unwrap(), i as f64);
628        }
629
630        drop(writer);
631        drop(stream);
632
633        let requests = test_consumer.requests.lock().unwrap();
634        assert_eq!(total_double_points(&requests, "networked_double"), 12);
635
636        let stats = stats.snapshot();
637        assert_eq!(stats.successful_requests as usize, requests.len());
638        assert_eq!(stats.simulated_failures, stats.successful_requests);
639        assert_eq!(stats.retries, stats.simulated_failures);
640        assert!(stats.attempts > stats.successful_requests);
641        assert!(stats.delivered_bytes > 0);
642        assert!(stats.simulated_sleep >= Duration::from_micros(stats.attempts * 10));
643    }
644
645    #[test_log::test]
646    fn simulated_network_all_requests_timeout_falls_back_under_backpressure() {
647        let primary_destination = Arc::new(TestDatasourceStream {
648            requests: Mutex::new(vec![]),
649        });
650        let fallback_destination = Arc::new(TestDatasourceStream {
651            requests: Mutex::new(vec![]),
652        });
653        let timeout = Duration::from_millis(5);
654        let simulated_network = SimulatedNetworkConsumer::new(
655            primary_destination.clone(),
656            SimulatedNetworkConfig::default()
657                .with_latency(timeout, Duration::ZERO)
658                .with_failure_pattern(SimulatedNetworkFailure::AllRequestsTimeout),
659        );
660        let stats = simulated_network.stats();
661        let stream = create_stream_with_consumer_and_options(
662            RequestConsumerWithFallback::new(simulated_network, fallback_destination.clone()),
663            NominalStreamOpts {
664                max_points_per_record: 2,
665                max_request_delay: Duration::from_millis(20),
666                max_buffered_requests: 1,
667                request_dispatcher_tasks: 1,
668                base_api_url: PRODUCTION_API_URL.to_string(),
669            },
670        );
671
672        let cd = ChannelDescriptor::new("timeout_double");
673        let mut writer = stream.double_writer(cd);
674        for i in 0..12 {
675            writer.push(UNIX_EPOCH.elapsed().unwrap(), i as f64);
676        }
677
678        drop(writer);
679        drop(stream);
680
681        let primary_requests = primary_destination.requests.lock().unwrap();
682        assert!(primary_requests.is_empty());
683
684        let fallback_requests = fallback_destination.requests.lock().unwrap();
685        assert_eq!(fallback_requests.len(), 6);
686        assert_eq!(
687            total_double_points(&fallback_requests, "timeout_double"),
688            12
689        );
690
691        let stats = stats.snapshot();
692        assert_eq!(stats.attempts, 6);
693        assert_eq!(stats.simulated_failures, 6);
694        assert_eq!(stats.successful_requests, 0);
695        assert_eq!(stats.retries, 0);
696        assert_eq!(stats.delivered_bytes, 0);
697        assert!(stats.simulated_sleep >= timeout * stats.attempts as u32);
698    }
699
700    #[test_log::test]
701    fn test_writer_types() {
702        let (test_consumer, stream) = create_test_stream();
703
704        let cd1 = ChannelDescriptor::new("double");
705        let cd2 = ChannelDescriptor::new("string");
706        let cd3 = ChannelDescriptor::new("int");
707        let cd4 = ChannelDescriptor::new("uint64");
708        let cd5 = ChannelDescriptor::new("struct");
709        let cd6 = ChannelDescriptor::new("double_array");
710        let cd7 = ChannelDescriptor::new("string_array");
711        let mut double_writer = stream.double_writer(cd1);
712        let mut string_writer = stream.string_writer(cd2);
713        let mut integer_writer = stream.integer_writer(cd3);
714        let mut uint64_writer = stream.uint64_writer(cd4);
715        let mut struct_writer = stream.struct_writer(cd5);
716        let mut double_array_writer = stream.double_array_writer(cd6);
717        let mut string_array_writer = stream.string_array_writer(cd7);
718
719        for i in 0..5000 {
720            let start_time = UNIX_EPOCH.elapsed().unwrap();
721            let value = i % 50;
722            double_writer.push(start_time, value as f64);
723            string_writer.push(start_time, format!("{}", value));
724            integer_writer.push(start_time, value);
725            uint64_writer.push(start_time, value as u64);
726            struct_writer.push(start_time, format!("{{\"v\":{}}}", value));
727            double_array_writer.push(start_time, vec![value as f64, (value + 1) as f64]);
728            string_array_writer.push(start_time, vec![format!("{}", value)]);
729        }
730
731        drop(double_writer);
732        drop(string_writer);
733        drop(integer_writer);
734        drop(uint64_writer);
735        drop(struct_writer);
736        drop(double_array_writer);
737        drop(string_array_writer);
738        drop(stream);
739
740        let requests = test_consumer.requests.lock().unwrap();
741        let counts = point_counts_by_channel(&requests);
742
743        assert_eq!(counts.get("double"), Some(&("double", 5000)));
744        assert_eq!(counts.get("string"), Some(&("string", 5000)));
745        assert_eq!(counts.get("int"), Some(&("int", 5000)));
746        assert_eq!(counts.get("uint64"), Some(&("uint64", 5000)));
747        assert_eq!(counts.get("struct"), Some(&("struct", 5000)));
748        assert_eq!(counts.get("double_array"), Some(&("double_array", 5000)));
749        assert_eq!(counts.get("string_array"), Some(&("string_array", 5000)));
750    }
751}