Module consumergroup 

Consumer groups for NATS

Allows creating Kafka consumer-group style sets of consumers for a JetStream stream. This class assumes that it has a monopoly on consuming the stream, i.e. that no other interface is used to consume it. Given a subject subject a group of ConsumerClients will consume all messages on the subject subject.[part].> for every part < partition_parts. partition_parts is defined in the ConsumerClientConfig. When a set of ConsumerClients are consuming a stream, it is guaranteed that messages belonging to the same part will be treated in order. But message consumption is not guaranteed to be globally ordered, as more consumers running in parallel can outpace one another. A group of ConsumerClients will try their best to have the same parts allocated to the same ConsumerClient over time. However, when a new ConsumerClient goes online or offline, all ConsumerClients will briefly stop reading the event stream, and repartition the stream among them before they start reading messages again, with no regard to the previous distribution of parts in the partition.

Example

This example connects to a stream, that may or may not exist, and prints all messages received. The cool part is, you can run multiple instances of this program, and they’ll distribute the work among themselves!

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
	let client = async_nats::connect("localhost:4242").await?;
	let jetstream = jetstream::new(client.clone());
	let stream = jetstream
		.get_or_create_stream(stream::Config {
			name: "eventstream".to_owned(),
			subjects: vec!["events.>".to_owned()],
			retention: stream::RetentionPolicy::WorkQueue,
			subject_transform: Some(SubjectTransform {
				source: "events.*.>".to_owned(),
				destination: "events.{{partition(100, 1)}}.>".to_owned(),
			}),
			..Default::default()
		})
		.await?;

	let mut client = consumergroup::ConsumerClient::new(
		consumergroup::ConsumerClientConfig {
			consumer_name: format!("worker_{}", process::id()),
			subject: "events".to_owned(),
			num_parts: 100,
			..Default::default()
		},
		client,
		stream,
	)
	.await?;

	loop {
		if let Some(message) = client.get_message().await? {
			println!("Received message {:?}", message);
		}
	}
	Ok(())
}

Structs

ConsumerClient

Handle controlling a single consumer.

ConsumerClientConfig

Configuration for a ConsumerClient.

Functions

update_or_recreate_consumer 🔒

Updates a consumer to give it a subset of the parts in the stream partition, compared to all other consumers currently registered in the stream. If the consumer no longer exists, possibly as a result of the inactive_threshold, the consumer is recreated.