use anyhow::anyhow;
use anyhow::Result;
use std::collections::HashSet;
use std::sync::Arc;
use tokio_nsq::{
NSQChannel, NSQConsumer, NSQConsumerConfig, NSQConsumerConfigSources, NSQConsumerLookupConfig,
NSQEvent, NSQProducer, NSQProducerConfig, NSQTopic,
};
pub async fn get_cusumer(topic: &str, ch: &str, addrs: HashSet<String>) -> NSQConsumer {
let topic = NSQTopic::new(topic).unwrap();
let channel = NSQChannel::new(ch).unwrap();
NSQConsumerConfig::new(topic, channel)
.set_max_in_flight(15)
.set_sources(NSQConsumerConfigSources::Lookup(
NSQConsumerLookupConfig::new().set_addresses(addrs),
))
.build()
}
pub async fn producer(addr: &str) -> Result<NSQProducer> {
let mut build = NSQProducerConfig::new(addr).build();
if is_healthy(build.consume().await.unwrap()) {
Ok(build)
} else {
Err(anyhow!("nsq is not healthy"))
}
}
pub fn is_healthy(evt: NSQEvent) -> bool {
match evt {
tokio_nsq::NSQEvent::Healthy() => true,
_ => false,
}
}
pub fn is_ok(evt: NSQEvent) -> bool {
match evt {
tokio_nsq::NSQEvent::Ok() => true,
_ => false,
}
}
pub fn is_unhealthy(evt: NSQEvent) -> bool {
match evt {
tokio_nsq::NSQEvent::Unhealthy() => true,
_ => false,
}
}
pub fn topic(tp: &str) -> Arc<NSQTopic> {
NSQTopic::new(tp).unwrap()
}