use std::borrow::Cow;
use std::collections::HashMap;
use std::num::NonZeroU32;
use std::ops::RangeInclusive;
use std::sync::{Mutex, PoisonError};
use std::time::Duration;
use ruststream::{BrokerMoves, DeclareRetryError, FromName, RetryDeclaration, SubscriptionSource};
use crate::broker::ConnectedPubSubBroker;
use crate::error::PubSubError;
use crate::subscriber::PubSubSubscriber;
const DEFAULT_BATCH_WAIT: Duration = Duration::from_millis(50);
const DELIVERY_ATTEMPTS: RangeInclusive<i32> = 5..=100;
const DEFAULT_MAX_LEASE: Duration = Duration::from_hours(1);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DeliveryLimits {
pub(crate) max_delivery_attempts: Option<i32>,
pub(crate) max_lease: Duration,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
pub struct GooglePubSub {
name: String,
create_with_topic: Option<String>,
max_outstanding: Option<i64>,
ack_extension: Option<Duration>,
batch_wait: Duration,
max_attempts: Option<NonZeroU32>,
dead_letter: Option<String>,
max_lease: Duration,
}
impl GooglePubSub {
pub fn new(name: impl Into<String>) -> Self {
Self {
name: name.into(),
create_with_topic: None,
max_outstanding: None,
ack_extension: None,
batch_wait: DEFAULT_BATCH_WAIT,
max_attempts: None,
dead_letter: None,
max_lease: DEFAULT_MAX_LEASE,
}
}
pub fn create_with_topic(mut self, topic: impl Into<String>) -> Self {
self.create_with_topic = Some(topic.into());
self
}
pub fn max_outstanding(mut self, messages: i64) -> Self {
self.max_outstanding = Some(messages);
self
}
pub fn ack_extension(mut self, extension: Duration) -> Self {
self.ack_extension = Some(extension);
self
}
pub fn max_lease(mut self, lease: Duration) -> Self {
self.max_lease = lease;
self
}
pub fn batch_wait(mut self, wait: Duration) -> Self {
self.batch_wait = wait;
self
}
#[must_use]
pub fn subscription(&self) -> &str {
&self.name
}
pub(crate) fn create_topic_ref(&self) -> Option<&str> {
self.create_with_topic.as_deref()
}
pub(crate) fn max_outstanding_value(&self) -> Option<i64> {
self.max_outstanding
}
pub(crate) fn ack_extension_value(&self) -> Option<Duration> {
self.ack_extension
}
pub(crate) fn batch_wait_value(&self) -> Duration {
self.batch_wait
}
pub(crate) fn limits(&self) -> DeliveryLimits {
DeliveryLimits {
max_delivery_attempts: self.dead_letter_policy().map(|(_, attempts)| attempts),
max_lease: self.max_lease,
}
}
fn declared_attempts(&self) -> Option<i32> {
self.max_attempts
.map(|attempts| i32::try_from(attempts.get()).unwrap_or(i32::MAX))
}
pub(crate) fn dead_letter_policy(&self) -> Option<(&str, i32)> {
Some((self.dead_letter.as_deref()?, self.declared_attempts()?))
}
fn take_declaration(mut self, declaration: &RetryDeclaration) -> Self {
self.max_attempts = declaration.max_attempts();
self.dead_letter = declaration.dead_letter().map(ToOwned::to_owned);
self
}
pub(crate) fn validate(&self) -> Result<(), PubSubError> {
if self.name.is_empty() {
return Err(PubSubError::InvalidDescriptor(
"subscription name must be non-empty".into(),
));
}
if self.create_with_topic.as_deref() == Some("") {
return Err(PubSubError::InvalidDescriptor(
"topic name must be non-empty".into(),
));
}
self.validate_declaration()
}
fn validate_declaration(&self) -> Result<(), PubSubError> {
match (self.declared_attempts(), self.dead_letter.as_deref()) {
(Some(attempts), Some(topic)) => {
if topic.is_empty() {
return Err(PubSubError::InvalidDescriptor(
"dead-letter topic must be non-empty".into(),
));
}
if !DELIVERY_ATTEMPTS.contains(&attempts) {
return Err(PubSubError::InvalidDescriptor(format!(
"max_attempts({attempts}) is outside the {}..={} a Pub/Sub dead-letter \
policy accepts",
DELIVERY_ATTEMPTS.start(),
DELIVERY_ATTEMPTS.end(),
)));
}
Ok(())
}
(Some(_), None) => Err(PubSubError::InvalidDescriptor(format!(
"subscription '{}' declares max_attempts without dead_letter; a Pub/Sub \
dead-letter policy needs the topic too",
self.name,
))),
(None, Some(_)) => Err(PubSubError::InvalidDescriptor(format!(
"subscription '{}' declares dead_letter without max_attempts; a Pub/Sub \
dead-letter policy needs the attempt count too",
self.name,
))),
(None, None) => Ok(()),
}
}
}
impl FromName for GooglePubSub {
fn from_name(name: impl Into<Cow<'static, str>>) -> Self {
Self::new(name.into().into_owned())
}
}
impl SubscriptionSource<ConnectedPubSubBroker> for GooglePubSub {
type Subscriber = PubSubSubscriber;
type Copies = BrokerMoves;
fn name(&self) -> &str {
self.subscription()
}
async fn subscribe(
self,
connected: &ConnectedPubSubBroker,
) -> Result<PubSubSubscriber, PubSubError> {
connected.subscribe_descriptor(self).await
}
fn declare_retry(self, declaration: &RetryDeclaration) -> Self {
self.take_declaration(declaration)
}
}
#[derive(Debug, Default)]
pub(crate) struct DeclaredRetries(Mutex<HashMap<String, RetryDeclaration>>);
impl DeclaredRetries {
pub(crate) fn take(
&self,
name: &str,
declaration: &RetryDeclaration,
) -> Result<(), DeclareRetryError> {
GooglePubSub::new(name)
.take_declaration(declaration)
.validate()
.map_err(refusal)?;
let conflicts = {
let mut taken = self.0.lock().unwrap_or_else(PoisonError::into_inner);
match taken.get(name) {
Some(already) if already != declaration => true,
_ => {
taken.insert(name.to_owned(), declaration.clone());
false
}
}
};
if conflicts {
return Err(refusal(PubSubError::InvalidDescriptor(format!(
"subscription '{name}' is mounted twice with different retry \
declarations, and one subscription carries one dead-letter policy",
))));
}
Ok(())
}
pub(crate) fn source(&self, name: &str) -> GooglePubSub {
let source = GooglePubSub::new(name);
match self
.0
.lock()
.unwrap_or_else(PoisonError::into_inner)
.get(name)
{
Some(declaration) => source.take_declaration(declaration),
None => source,
}
}
}
fn refusal(err: PubSubError) -> DeclareRetryError {
DeclareRetryError::Broker(Box::new(err))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn empty_subscription_name_is_rejected_before_io() {
assert!(matches!(
GooglePubSub::new("").validate(),
Err(PubSubError::InvalidDescriptor(_))
));
}
#[test]
fn empty_topic_name_is_rejected_before_io() {
assert!(matches!(
GooglePubSub::new("s").create_with_topic("").validate(),
Err(PubSubError::InvalidDescriptor(_))
));
}
fn declaration(attempts: Option<u32>, destination: Option<&str>) -> RetryDeclaration {
let mut declaration = RetryDeclaration::new();
if let Some(attempts) = attempts {
declaration =
declaration.with_max_attempts(NonZeroU32::new(attempts).expect("non-zero"));
}
if let Some(destination) = destination {
declaration = declaration.with_dead_letter(destination.to_owned());
}
declaration
}
fn declared(attempts: Option<u32>, destination: Option<&str>) -> GooglePubSub {
SubscriptionSource::<ConnectedPubSubBroker>::declare_retry(
GooglePubSub::new("orders-workers"),
&declaration(attempts, destination),
)
}
#[test]
fn a_full_declaration_becomes_the_subscriptions_dead_letter_policy() {
let source = declared(Some(5), Some("orders-dead"));
assert_eq!(source.dead_letter_policy(), Some(("orders-dead", 5)));
}
#[test]
fn an_empty_declaration_leaves_the_subscription_untouched() {
let source = declared(None, None);
assert_eq!(source.dead_letter_policy(), None);
assert!(source.validate().is_ok());
}
#[test]
fn half_a_declaration_is_refused_before_io() {
for (attempts, destination) in [(Some(5), None), (None, Some("orders-dead"))] {
let source = declared(attempts, destination);
assert!(
matches!(source.validate(), Err(PubSubError::InvalidDescriptor(_))),
"declaring {attempts:?} / {destination:?} must be refused",
);
}
}
#[test]
fn a_cap_outside_the_services_range_is_refused_before_io() {
for attempts in [1, 4, 101, 1_000] {
let source = declared(Some(attempts), Some("orders-dead"));
assert!(
matches!(source.validate(), Err(PubSubError::InvalidDescriptor(_))),
"max_attempts({attempts}) must be refused",
);
}
for attempts in [5, 100] {
let source = declared(Some(attempts), Some("orders-dead"));
assert!(
source.validate().is_ok(),
"max_attempts({attempts}) must be accepted",
);
}
}
#[test]
fn a_cap_past_the_wire_type_is_refused_rather_than_reduced() {
let source = declared(Some(u32::MAX), Some("orders-dead"));
assert_eq!(source.declared_attempts(), Some(i32::MAX));
assert!(matches!(
source.validate(),
Err(PubSubError::InvalidDescriptor(_))
));
}
#[test]
fn an_empty_dead_letter_topic_is_refused_before_io() {
let source = declared(Some(5), Some(""));
assert!(matches!(
source.validate(),
Err(PubSubError::InvalidDescriptor(_))
));
}
#[test]
fn a_declaration_for_a_bare_name_reaches_the_subscription_it_opens() {
let taken = DeclaredRetries::default();
taken
.take("orders-workers", &declaration(Some(5), Some("orders-dead")))
.expect("a full declaration is one a dead-letter policy carries");
assert_eq!(
taken.source("orders-workers").dead_letter_policy(),
Some(("orders-dead", 5)),
);
}
#[test]
fn a_bare_name_with_no_declaration_opens_the_subscription_untouched() {
let taken = DeclaredRetries::default();
taken
.take("orders-workers", &RetryDeclaration::new())
.expect("a registration may declare nothing");
assert_eq!(
taken.source("orders-workers"),
GooglePubSub::new("orders-workers")
);
}
#[test]
fn a_bare_name_is_refused_what_a_dead_letter_policy_cannot_express() {
let taken = DeclaredRetries::default();
for (attempts, destination) in [
(Some(5), None),
(None, Some("orders-dead")),
(Some(3), Some("orders-dead")),
] {
let err = taken
.take("orders-workers", &declaration(attempts, destination))
.expect_err("the declaration must be refused");
assert!(
matches!(err, DeclareRetryError::Broker(_)),
"declaring {attempts:?} / {destination:?} must refuse as a broker rejection",
);
}
}
#[test]
fn one_subscription_declared_twice_differently_is_refused() {
let taken = DeclaredRetries::default();
taken
.take("orders-workers", &declaration(Some(5), Some("orders-dead")))
.expect("the first declaration is taken");
let err = taken
.take(
"orders-workers",
&declaration(Some(10), Some("orders-dead")),
)
.expect_err("a contradicting declaration must be refused");
assert!(matches!(err, DeclareRetryError::Broker(_)));
assert_eq!(
taken.source("orders-workers").dead_letter_policy(),
Some(("orders-dead", 5)),
"the refusal leaves the declaration already taken alone",
);
}
#[test]
fn one_subscription_declared_twice_alike_is_taken_once() {
let taken = DeclaredRetries::default();
let declared = declaration(Some(5), Some("orders-dead"));
taken.take("orders-workers", &declared).expect("the first");
taken.take("orders-workers", &declared).expect("the second");
assert_eq!(
taken.source("orders-workers").dead_letter_policy(),
Some(("orders-dead", 5)),
);
}
}