Cargo.toml

[package]
name = "shove"
version = "0.12.1"
description = "Async tasks via pubsub on steroids. Comes with built-in support for complex queue configurations, audit logs, autoscaling consumer groups and more."
edition = "2024"
rust-version = "1.91"
authors = ["Zannis Kalampoukis <zkalampoukis@gmail.com>"]
license = "MIT"
repository = "https://github.com/zannis/shove"
keywords = ["messaging", "rabbitmq", "pubsub", "consumer-groups", "autoscaling"]
categories = ["asynchronous"]

[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]

exclude = [
  "docs/",
  "package.json",
  "package-lock.json",
  "tsconfig.json",
  "vocs.config.ts",
  "wrangler.toml",
  ".github/workflows/docs-build.yml",
  ".github/workflows/docs-preview.yml",
  "node_modules/",
]

[lints.clippy]
absolute-paths = "deny"

[features]
default = []

# RabbitMQ Pubsub
rabbitmq = [
  "dep:lapin",
  "dep:futures-lite",
  "dep:reqwest",
  "dep:arc-swap",
  "dep:url",
]

# Opt-in exactly-once delivery for RabbitMQ consumers via AMQP transactions.
# Makes publish-to-hold-queue + ack atomic (tx_select/tx_commit), eliminating
# the publish-then-ack race at the cost of ~10-15× lower throughput per channel.
rabbitmq-transactional = ["rabbitmq"]

# SNS Publisher
pub-aws-sns = ["dep:aws-config", "dep:aws-sdk-sns"]

# Complete pubsub with AWS SNS and AWS SQS
aws-sns-sqs = ["pub-aws-sns", "dep:aws-sdk-sqs"]

# Audit traits and default audit topic
audit = []

# NATS pubsub
nats = ["dep:async-nats", "dep:futures-util"]

# Apache Kafka pubsub
kafka = ["dep:rdkafka", "dep:futures-util"]

# TLS + SASL support for Kafka. Pulls in rdkafka's `ssl` (OpenSSL) and `sasl`
# (Cyrus SASL) features. For GSSAPI/Kerberos or a vendored OpenSSL, users can
# additionally enable `rdkafka/gssapi` or `rdkafka/ssl-vendored` downstream.
kafka-ssl = ["kafka", "rdkafka?/ssl", "rdkafka?/sasl"]

# AWS MSK IAM authentication. Adds the `MskIam` variant to `KafkaSasl`,
# which uses `SASL_SSL` + `OAUTHBEARER` with auto-refreshed presigned tokens
# signed against `kafka-cluster:Connect`. Implies TLS.
kafka-msk-iam = [
  "kafka-ssl",
  "dep:aws-config",
  "dep:aws-credential-types",
  "dep:aws-msk-iam-sasl-signer",
]

# Confluent Schema Registry decode for Kafka consumers. Adds an async, cached
# registry client and an opt-in decode stage that strips the Confluent wire
# frame and validates schema identity before delegating to the topic's codec.
kafka-schema-registry = [
  "kafka",
  "dep:reqwest",
  "dep:dashmap",
  "dep:futures-util",
]

# In-process backend
inmemory = ["dep:dashmap"]

# Redis Streams pubsub
redis-streams = ["dep:redis"]

# Emit operational metrics via the `metrics` facade. Consumer services
# install their own recorder (e.g. `metrics-exporter-prometheus`) and
# expose the endpoint themselves; this crate never opens a port.
metrics = ["dep:metrics"]

# Protobuf codec via `prost`. Enables `ProtobufCodec<M>` for any
# `M: prost::Message + Default`.
protobuf = ["dep:prost"]

# Simple Binary Encoding (SBE) codec. Zero-copy: topics carry `SbeFrame<T>`
# wire frames that publish and decode without re-encoding or copying, so
# handlers wrap sbe-tool-generated flyweight decoders directly over the
# received buffer. No extra dependencies.
sbe = []

# Test-only seams for integration tests (not part of the public API for normal
# builds). Currently exposes the Kafka admin OAUTHBEARER priming path so the
# OAUTHBEARER integration test can drive it against a local broker with an
# unsecured JWT (MSK's presigned-URL tokens can't be validated without AWS).
test-support = []

[dependencies]
serde = { version = "1", features = ["derive", "rc"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }
thiserror = "2"
tracing = "0.1"
tokio-util = { version = "0.7", features = ["rt"] }
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }
lapin = { version = "4", features = ["rustls--ring"], optional = true }
futures-lite = { version = "2", optional = true }
reqwest = { version = "0.13", features = ["json"], optional = true }
aws-config = { version = "1", features = [
  "behavior-version-latest",
  "default-https-client",
  "rt-tokio",
], default-features = false, optional = true }
aws-credential-types = { version = "1", optional = true }
aws-msk-iam-sasl-signer = { version = "1", optional = true }
aws-sdk-sns = { version = "1", features = [
  "behavior-version-latest",
  "default-https-client",
  "rt-tokio",
], default-features = false, optional = true }
aws-sdk-sqs = { version = "1", features = [
  "behavior-version-latest",
  "default-https-client",
  "rt-tokio",
], default-features = false, optional = true }
rand = "0.10"
arc-swap = { version = "1", optional = true }
async-nats = { version = "0.49", optional = true }
bytes = "1"
futures-util = { version = "0.3", optional = true }
rdkafka = { version = "0.39", features = ["tokio", "tracing"], optional = true }
dashmap = { version = "6", optional = true }
redis = { version = "1", features = [
  "tokio-comp",
  "streams",
  "tokio-rustls-comp",
  "cluster-async",
], optional = true }
metrics = { version = "0.24", optional = true }
url = { version = "2", optional = true }
prost = { version = "0.14", optional = true }

[target.'cfg(windows)'.dependencies]
rdkafka = { version = "0.39", features = [
  "tokio",
  "tracing",
  "cmake-build",
], optional = true }

[dev-dependencies]
criterion = { version = "0.8", features = ["async_tokio"] }
testcontainers = "0.27"
testcontainers-modules = { version = "0.15", features = [
  "rabbitmq",
  "localstack",
  "nats",
  "kafka",
  "redis",
] }
# RUSTSEC-2026-0145: testcontainers 0.27.3 pulls in astral-tokio-tar 0.6.1 which
# has a known vulnerability. Pin to 0.6.2 until testcontainers ships a fix.
astral-tokio-tar = "0.6.2"
tokio = { version = "1", features = ["full", "test-util"] }
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
clap = { version = "4", features = ["derive"] }
rand = "0.10"
metrics-util = "0.20"
libc = "0.2"
axum = "0.8"
[target.'cfg(target_os = "macos")'.dev-dependencies]
mach2 = "0.6"

[[bench]]
name = "autoscaler"
harness = false
required-features = ["rabbitmq"]

[[bench]]
name = "consumer_overhead"
harness = false
required-features = ["rabbitmq"]

[[bench]]
name = "publish_throughput"
harness = false
required-features = ["rabbitmq"]

[[example]]
name = "rabbitmq_basic_pubsub"
path = "examples/rabbitmq/basic_pubsub.rs"
required-features = ["rabbitmq"]

[[example]]
name = "rabbitmq_sequenced_pubsub"
path = "examples/rabbitmq/sequenced_pubsub.rs"
required-features = ["rabbitmq"]

[[example]]
name = "rabbitmq_consumer_groups"
path = "examples/rabbitmq/consumer_groups.rs"
required-features = ["rabbitmq"]

[[example]]
name = "rabbitmq_audited_consumer"
path = "examples/rabbitmq/audited_consumer.rs"
required-features = ["rabbitmq", "audit"]

[[example]]
name = "rabbitmq_concurrent_pubsub"
path = "examples/rabbitmq/concurrent_pubsub.rs"
required-features = ["rabbitmq"]

[[example]]
name = "rabbitmq_stress"
path = "examples/rabbitmq/stress.rs"
required-features = ["rabbitmq"]

[[example]]
name = "sqs_basic_pubsub"
path = "examples/sqs/basic_pubsub.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "sqs_sequenced_pubsub"
path = "examples/sqs/sequenced_pubsub.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "sqs_concurrent_pubsub"
path = "examples/sqs/concurrent_pubsub.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "sqs_consumer_groups"
path = "examples/sqs/consumer_groups.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "sqs_audited_consumer"
path = "examples/sqs/audited_consumer.rs"
required-features = ["aws-sns-sqs", "audit"]

[[example]]
name = "sqs_autoscaler"
path = "examples/sqs/autoscaler.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "sqs_stress"
path = "examples/sqs/stress.rs"
required-features = ["aws-sns-sqs"]

[[example]]
name = "rabbitmq_exactly_once"
path = "examples/rabbitmq/exactly_once.rs"
required-features = ["rabbitmq-transactional"]

[[example]]
name = "nats_basic"
path = "examples/nats/basic.rs"
required-features = ["nats"]

[[example]]
name = "nats_sequenced"
path = "examples/nats/sequenced.rs"
required-features = ["nats"]

[[example]]
name = "nats_stress"
path = "examples/nats/stress.rs"
required-features = ["nats"]

[[example]]
name = "nats_audited_consumer"
path = "examples/nats/audited_consumer.rs"
required-features = ["nats", "audit"]

[[example]]
name = "nats_raw_bytes"
path = "examples/nats/raw_bytes.rs"
required-features = ["nats"]

[[example]]
name = "kafka_basic"
path = "examples/kafka/basic.rs"
required-features = ["kafka"]

[[example]]
name = "kafka_sequenced"
path = "examples/kafka/sequenced.rs"
required-features = ["kafka"]

[[example]]
name = "kafka_stress"
path = "examples/kafka/stress.rs"
required-features = ["kafka"]

[[example]]
name = "kafka_audited_consumer"
path = "examples/kafka/audited_consumer.rs"
required-features = ["kafka", "audit"]

[[example]]
name = "kafka_msk_iam"
path = "examples/kafka/msk_iam.rs"
required-features = ["kafka-msk-iam"]

[[example]]
name = "kafka_protobuf_pubsub"
path = "examples/kafka/protobuf_pubsub.rs"
required-features = ["kafka", "protobuf"]

[[example]]
name = "inmemory_basic"
path = "examples/inmemory/basic.rs"
required-features = ["inmemory"]

[[example]]
name = "inmemory_sequenced"
path = "examples/inmemory/sequenced.rs"
required-features = ["inmemory"]

[[example]]
name = "inmemory_consumer_groups"
path = "examples/inmemory/consumer_groups.rs"
required-features = ["inmemory"]

[[example]]
name = "inmemory_audited_consumer"
path = "examples/inmemory/audited_consumer.rs"
required-features = ["inmemory", "audit"]

[[example]]
name = "inmemory_stress"
path = "examples/inmemory/stress.rs"
required-features = ["inmemory"]


[[example]]
name = "redis_basic"
path = "examples/redis/basic.rs"
required-features = ["redis-streams"]

[[example]]
name = "redis_sequenced"
path = "examples/redis/sequenced.rs"
required-features = ["redis-streams"]

[[example]]
name = "redis_stress"
path = "examples/redis/stress.rs"
required-features = ["redis-streams"]

[[example]]
name = "redis_consumer_groups"
path = "examples/redis/consumer_groups.rs"
required-features = ["redis-streams"]

[[example]]
name = "redis_autoscaler"
path = "examples/redis/autoscaler.rs"
required-features = ["redis-streams"]

[[example]]
name = "redis_audited_consumer"
path = "examples/redis/audited_consumer.rs"
required-features = ["redis-streams", "audit"]