rivetkit 2.3.0-rc.12

Rust SDK for RivetKit actors, actions, events, queues, and test harnesses
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

mod moved_tests {
	use futures::StreamExt;
	use tokio_util::sync::CancellationToken;

	use crate::queue::{QueueStreamExt, QueueStreamOpts};
	use rivetkit_core::{ActorContext, Kv};

	#[tokio::test]
	async fn queue_stream_yields_messages_through_stream_ext_combinators() {
		let ctx = ActorContext::new_with_kv(
			"actor-id",
			"test",
			Vec::new(),
			"local",
			Kv::new_in_memory(),
		);
		let queue = ctx.queue();

		queue.send("alpha", br#"{"value":1}"#).await.expect("send alpha");
		queue.send("beta", br#"{"value":2}"#).await.expect("send beta");

		let names = queue
			.stream(QueueStreamOpts::default())
			.map(|message| message.name)
			.take(2)
			.collect::<Vec<_>>()
			.await;

		assert_eq!(names, vec!["alpha".to_owned(), "beta".to_owned()]);
	}

	#[tokio::test]
	async fn queue_stream_honors_name_filters() {
		let ctx = ActorContext::new_with_kv(
			"actor-id",
			"test",
			Vec::new(),
			"local",
			Kv::new_in_memory(),
		);
		let queue = ctx.queue();

		queue.send("skip", b"1").await.expect("send skip");
		queue.send("keep", b"2").await.expect("send keep");

		let message = queue
			.stream(QueueStreamOpts {
				names: Some(vec!["keep".to_owned()]),
				signal: None,
			})
			.next()
			.await
			.expect("filtered stream should yield");

		assert_eq!(message.name, "keep");
		assert_eq!(message.body, b"2");
	}

	#[tokio::test]
	async fn queue_stream_ends_when_cancellation_signal_is_already_fired() {
		let ctx = ActorContext::new_with_kv(
			"actor-id",
			"test",
			Vec::new(),
			"local",
			Kv::new_in_memory(),
		);
		let queue = ctx.queue();
		let signal = CancellationToken::new();
		signal.cancel();

		let next = queue
			.stream(QueueStreamOpts {
				names: None,
				signal: Some(signal),
			})
			.next()
			.await;

		assert!(next.is_none());
	}
}