amqpsy 0.1.0

Extremely opinionated AMQP PubSub library
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

<h1 align="center">amqpsy</h1>
<div align="center">
 <strong>
     Extremely opinionated AMQP PubSub library
 </strong>
</div>

<br />

<div align="center">
  <!-- Crates version -->
  <a href="https://crates.io/crates/amqpsy">
    <img src="https://img.shields.io/crates/v/amqpsy.svg?style=flat-square"
    alt="Crates.io version" />
  </a>
  <!-- Downloads -->
  <a href="https://crates.io/crates/amqpsy">
    <img src="https://img.shields.io/crates/d/amqpsy.svg?style=flat-square"
      alt="Download" />
  </a>
  <!-- docs.rs docs -->
  <a href="https://docs.rs/amqpsy">
    <img src="https://img.shields.io/badge/docs-latest-blue.svg?style=flat-square"
      alt="docs.rs docs" />
  </a>
</div>
<br/>

Built on top of `lapin` - this library is designed to be a simple and easy to use AMQP PubSub library.
It's extremely opinionated and comes with some batteries* included.

## Features
- Publishers with default configurations for Events and Commands.
- Consumers have DLQ enabled by default.
- Fluent interface for consumer groups.
- Ergonomic error handling API: retry, DLQ, or invalid message options.
- Enforces Protobuf for messages, ensuring backward compatibility.
- Distributed tracing enabled with Otel: linked traces from publisher to consumer.
- Chaos Monkey: simulate duplicate messages, publish failure etc.

## Usage

### Consumer Groups
```rust,ignore
    let context = AppContext::new(); // your app context
    let config = amqpsy::AmqpConfig {
        connection_string: "<your connection string>".to_string(),
    };

    #[rustfmt::skip]
    ConsumerGroup::builder(
        "app_name",
        config,
        context,
    )
    .for_topic_exchange("ledger")
        // Using default settings
        .consume(handlers::CreateOutboundHandler).await?
        // Using custom settings
        .consume_with_config(handlers::OutboundAcceptedHandler, ConsumerConfig::default().with_worker_count(5)).await?
    .then()
    .for_topic_exchange("provider")
        .consume(handlers::provider::InboundTransactionSettledHandler).await?
        .consume(handlers::provider::OutboundAcceptedHandler).await?
        .consume(handlers::provider::OutboundSettledHandler).await?
        .consume(handlers::provider::OutboundFailedHandler).await?
    .run_until_shutdown().await?; // This will block until shutdown

    // Example handler
    #[derive(Clone, Debug)]
    pub struct CreateOutboundHandler;

    impl AmqpHandler for CreateOutboundHandler {
        type Message = CreateOutboundTransaction; // Must be a Protobuffer Message
        type Context = AppContext;

        async fn handle(
            &self,
            context: Self::Context,
            message: Self::Message,
        ) -> Result<(), ConsumerError> {
            // Do something with the message

            // Or return error to nack the message
            // return Err(ConsumerError::Retry(...) // means the message retried before sent to DLQ
            // return Err(ConsumerError::Fatal(..)) // means the message sent to DLQ
            // return Err(ConsumerError::Invalid(..)) // means the message is dropped
            //
            // Also use the fluent API to handle errors
            //
            // context
            //      .db.add_transaction(&message).await
            //      .or_transient_error()?; // Other Options: or_fatal_error()? or_invalid_error()?

            Ok(()) // means the message is Ack'd
        }
    }

```

## Publishers

Pick one of three option using the `AmqpPublisher::new_*` methods.
* Command: Publisher confirm and mandatory routing enabled
* Event: Publisher confirm enabled but no mandatory routing enabled
* Event without Publisher confirmation: No publisher confirm and no mandatory routing

```rust,ignore
// commands::CreateOutboundTransaction is a Proto message
let command =
    AmqpPublisher::<commands::CreateOutboundTransaction>::new_for_command(
        config,
        "exchange_name",
    )
    .await?;
let event = AmqpPublisher::<events::OutboundTransactionSettled>::new_for_event(
    config,
    "exchange_name",
)
.await?; // or new_for_event_without_publisher_confirmation

// Publish a command
// Command has publisher confirm and mandatory routing enabled
command.publish(&command).await?;

// Publish an event
// Event has publisher confirm enabled but no mandatory routing enabled
event.publish(&event).await?;
```


## Distributed Tracing

Traces are linked between publisher and consumer with necessary tags set on them (prefixed with `amqpsy.*`)

![example 1](assets/tracing_example_1.png)

![example 2](assets/tracing_example_2.png)

## Chaos

Simulate duplicate mesages, publish failure etc. in your system.

Example shows a system with 10% duplicate messages and 10% publish failure.
![chaos example 2](assets/chaos_1.png)

* Enable feature `chaos`
* Specify one of the chaos parameteres as environment variables.
  * `AMQPSY_CHAOS_PUBLISHER_DUPLICATE_PERCENTAGE`: % of publish will be sent more than once to the broker.
  * `AMQPSY_CHAOS_PUBLISHER_FAILURE_PERCENTAGE`: % of publish will fail due to random error.

If you enable `chaos` feature but don't specify any parameter - the application will crash at startup.