Struct RabbitMqClientHandler 

pub struct RabbitMqClientHandler<BasicConsumerType: AsyncConsumer + Clone + Send + 'static> { /* private fields */ }

This struct is holding the state of the RabbitMQ client. The whole resiliency revolves around it. It will handle connection errors and backup the list of consumers (basic.consume) you declare and recreate them when something happens. If you need to publish to RabbitMQ as well, a PublishChannel will be extracted from it and regenerated if needed.

Implementations

impl<BasicConsumerType: AsyncConsumer + Clone + Send + 'static> RabbitMqClientHandler<BasicConsumerType>

pub async fn new( open_connection_arguments: OpenConnectionArguments, ) -> RabbitMqClientHandler<BasicConsumerType>

Examples found in repository

examples/one_consumer_no_publisher.rs (line 36)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a consumer on an 'incoming' queue and include it in the RabbitMqClientHandler
    let basic_consumer_args = ConsumerArguments::new("incoming", "my_first_resilient_consumer");
    let basic_consumer = MyRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(basic_consumer, basic_consumer_args)
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}

examples/only_a_publisher.rs (line 31)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_client: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a PublishChannel
    let mut publisher = resilient_rabbitmq_client.create_publisher().await.unwrap();
    // Get a channel from it. This channel is meant to be used to communicate with the PublishChannel type.
    let publish_sender = resilient_rabbitmq_client
        .get_publish_sender_clone()
        .unwrap();

    // For the example, we spawn a task which will publish some random data on an "incoming" routing key every second.
    tokio::task::spawn(async move {
        loop {
            publish_sender
                .send(DataToPublish::from(
                    "amq.direct".to_string(),
                    "incoming".to_string(),
                    "random data".to_string(),
                ))
                .await
                .unwrap();

            sleep(Duration::from_secs(1)).await;
        }
    });

    // Keep alive both the client and the publisher
    tokio::join!(
        publisher.keep_ready_to_publish(),
        resilient_rabbitmq_client.keep_alive()
    );
}

examples/multiple_kinds_of_consumers.rs (line 93)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<RabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // First kind of client : consumes from a specific queue a given data format
    let basic_consumer_first_kind_args =
        ConsumerArguments::new("first_queue", "first_kind_of_consumer").finish();
    let basic_consumer_first_kind = FirstKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::FirstKind(basic_consumer_first_kind),
            basic_consumer_first_kind_args,
        )
        .await;

    // Second kind of client : consumes from another queue another kind of data
    let basic_consumer_second_kind_args =
        ConsumerArguments::new("second_queue", "second_kind_of_consumer").finish();
    let basic_consumer_second_kind = SecondKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::SecondKind(basic_consumer_second_kind),
            basic_consumer_second_kind_args,
        )
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}

pub async fn try_reconnect(&mut self) -> Result<(), String>

pub async fn create_publisher(&mut self) -> Result<PublishChannel, String>

Examples found in repository

examples/only_a_publisher.rs (line 34)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_client: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a PublishChannel
    let mut publisher = resilient_rabbitmq_client.create_publisher().await.unwrap();
    // Get a channel from it. This channel is meant to be used to communicate with the PublishChannel type.
    let publish_sender = resilient_rabbitmq_client
        .get_publish_sender_clone()
        .unwrap();

    // For the example, we spawn a task which will publish some random data on an "incoming" routing key every second.
    tokio::task::spawn(async move {
        loop {
            publish_sender
                .send(DataToPublish::from(
                    "amq.direct".to_string(),
                    "incoming".to_string(),
                    "random data".to_string(),
                ))
                .await
                .unwrap();

            sleep(Duration::from_secs(1)).await;
        }
    });

    // Keep alive both the client and the publisher
    tokio::join!(
        publisher.keep_ready_to_publish(),
        resilient_rabbitmq_client.keep_alive()
    );
}

pub async fn new_consumer( &mut self, basic_consumer: BasicConsumerType, basic_consumer_args: BasicConsumeArguments, )

Examples found in repository

examples/one_consumer_no_publisher.rs (line 42)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a consumer on an 'incoming' queue and include it in the RabbitMqClientHandler
    let basic_consumer_args = ConsumerArguments::new("incoming", "my_first_resilient_consumer");
    let basic_consumer = MyRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(basic_consumer, basic_consumer_args)
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}

examples/multiple_kinds_of_consumers.rs (lines 100-103)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<RabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // First kind of client : consumes from a specific queue a given data format
    let basic_consumer_first_kind_args =
        ConsumerArguments::new("first_queue", "first_kind_of_consumer").finish();
    let basic_consumer_first_kind = FirstKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::FirstKind(basic_consumer_first_kind),
            basic_consumer_first_kind_args,
        )
        .await;

    // Second kind of client : consumes from another queue another kind of data
    let basic_consumer_second_kind_args =
        ConsumerArguments::new("second_queue", "second_kind_of_consumer").finish();
    let basic_consumer_second_kind = SecondKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::SecondKind(basic_consumer_second_kind),
            basic_consumer_second_kind_args,
        )
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}

pub async fn create_consumer( &mut self, basic_consumer: BasicConsumerType, basic_consumer_args: BasicConsumeArguments, ) -> Result<(), Error>

pub fn get_publish_sender_clone(&self) -> Result<Sender<DataToPublish>, String>

Examples found in repository

examples/only_a_publisher.rs (line 37)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_client: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a PublishChannel
    let mut publisher = resilient_rabbitmq_client.create_publisher().await.unwrap();
    // Get a channel from it. This channel is meant to be used to communicate with the PublishChannel type.
    let publish_sender = resilient_rabbitmq_client
        .get_publish_sender_clone()
        .unwrap();

    // For the example, we spawn a task which will publish some random data on an "incoming" routing key every second.
    tokio::task::spawn(async move {
        loop {
            publish_sender
                .send(DataToPublish::from(
                    "amq.direct".to_string(),
                    "incoming".to_string(),
                    "random data".to_string(),
                ))
                .await
                .unwrap();

            sleep(Duration::from_secs(1)).await;
        }
    });

    // Keep alive both the client and the publisher
    tokio::join!(
        publisher.keep_ready_to_publish(),
        resilient_rabbitmq_client.keep_alive()
    );
}

pub async fn keep_alive(&mut self)

Examples found in repository

examples/one_consumer_no_publisher.rs (line 46)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a consumer on an 'incoming' queue and include it in the RabbitMqClientHandler
    let basic_consumer_args = ConsumerArguments::new("incoming", "my_first_resilient_consumer");
    let basic_consumer = MyRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(basic_consumer, basic_consumer_args)
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}

examples/only_a_publisher.rs (line 59)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_client: RabbitMqClientHandler<MyRabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // Declare a PublishChannel
    let mut publisher = resilient_rabbitmq_client.create_publisher().await.unwrap();
    // Get a channel from it. This channel is meant to be used to communicate with the PublishChannel type.
    let publish_sender = resilient_rabbitmq_client
        .get_publish_sender_clone()
        .unwrap();

    // For the example, we spawn a task which will publish some random data on an "incoming" routing key every second.
    tokio::task::spawn(async move {
        loop {
            publish_sender
                .send(DataToPublish::from(
                    "amq.direct".to_string(),
                    "incoming".to_string(),
                    "random data".to_string(),
                ))
                .await
                .unwrap();

            sleep(Duration::from_secs(1)).await;
        }
    });

    // Keep alive both the client and the publisher
    tokio::join!(
        publisher.keep_ready_to_publish(),
        resilient_rabbitmq_client.keep_alive()
    );
}

examples/multiple_kinds_of_consumers.rs (line 118)

async fn main() {
    // Declare a RabbitMqClientHandler
    let rabbitmq_connection_arguments =
        ConnectionArguments::new("localhost", 5672, "guest", "guest");
    let mut resilient_rabbitmq_connection: RabbitMqClientHandler<RabbitMQClient> =
        RabbitMqClientHandler::new(rabbitmq_connection_arguments).await;

    // First kind of client : consumes from a specific queue a given data format
    let basic_consumer_first_kind_args =
        ConsumerArguments::new("first_queue", "first_kind_of_consumer").finish();
    let basic_consumer_first_kind = FirstKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::FirstKind(basic_consumer_first_kind),
            basic_consumer_first_kind_args,
        )
        .await;

    // Second kind of client : consumes from another queue another kind of data
    let basic_consumer_second_kind_args =
        ConsumerArguments::new("second_queue", "second_kind_of_consumer").finish();
    let basic_consumer_second_kind = SecondKindOfRabbitMQClient;
    resilient_rabbitmq_connection
        .new_consumer(
            RabbitMQClient::SecondKind(basic_consumer_second_kind),
            basic_consumer_second_kind_args,
        )
        .await;

    // Keep alive the RabbitMqClientHandler
    resilient_rabbitmq_connection.keep_alive().await;
}