rskit-messaging-rabbitmq 0.2.0-alpha.3

RabbitMQ adapter for rskit-messaging
Documentation 
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

use std::sync::Arc;

use rskit_errors::AppResult;
use rskit_messaging::{
    BrokerConfigExt, MessageConsumer, MessageProducer, MessagingFactory, MessagingRegistry,
};

use crate::Config;
use crate::consumer::RabbitMqConsumer;
use crate::producer::RabbitMqProducer;

/// Register `RabbitMQ` producer and consumer factories for `Vec<u8>` payloads.
pub fn register(registry: &mut MessagingRegistry<Vec<u8>>, config: Config) -> AppResult<()> {
    config.validate()?;
    if !config.base.enabled {
        return Ok(());
    }
    let adapter = config.base.adapter.clone();
    registry.register_backend(adapter, Arc::new(RabbitMqFactory { config }))
}

pub(crate) struct RabbitMqFactory {
    pub(crate) config: Config,
}

impl MessagingFactory<Vec<u8>> for RabbitMqFactory {
    fn create_producer(
        &self,
        _config: &rskit_messaging::BrokerConfig,
    ) -> AppResult<Arc<dyn MessageProducer<Vec<u8>>>> {
        Ok(Arc::new(RabbitMqProducer::new(self.config.clone())?))
    }

    fn create_consumer(
        &self,
        _config: &rskit_messaging::BrokerConfig,
    ) -> AppResult<Arc<dyn MessageConsumer<Vec<u8>>>> {
        Ok(Arc::new(RabbitMqConsumer::new(self.config.clone())?))
    }
}