esrs 0.18.0

A Prima.it-opinionated library to achieve cqrs/es
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

use chrono::Utc;
use futures::TryStreamExt;
use lapin::options::{BasicAckOptions, BasicConsumeOptions, QueueBindOptions, QueueDeclareOptions};
use lapin::types::FieldTable;
use lapin::{Connection, ConnectionProperties, Consumer, ExchangeKind};
use rand::prelude::IteratorRandom;
use uuid::Uuid;

use esrs::bus::rabbit::{RabbitEventBus, RabbitEventBusConfig};
use esrs::bus::EventBus;
use esrs::store::StoreEvent;

use crate::aggregate::{TestAggregate, TestEvent};

#[tokio::test]
async fn rabbit_event_bus_test() {
    let rabbit_url: String = std::env::var("RABBIT_URL").unwrap();

    let exchange: String = format!("{}_test_exchange", random_letters());
    let queue: String = format!("{}_test_queue", random_letters());
    let routing_key: String = format!("{}_test_routing_key", random_letters());

    let config: RabbitEventBusConfig = RabbitEventBusConfig::builder()
        .url(rabbit_url.as_str())
        .exchange(exchange.as_str())
        .exchange_kind(ExchangeKind::Fanout)
        .publish_routing_key(Some(routing_key.to_string()))
        .error_handler(Box::new(|error| panic!("{:?}", error)))
        .build();

    let bus: RabbitEventBus<TestAggregate> = match RabbitEventBus::new(config).await {
        Ok(bus) => bus,
        Err(error) => panic!("{:?}", error),
    };

    let mut consumer: Consumer = consumer(
        rabbit_url.as_str(),
        exchange.as_str(),
        queue.as_str(),
        routing_key.as_str(),
    )
    .await;

    let event_id: Uuid = Uuid::new_v4();
    let store_event: StoreEvent<TestEvent> = StoreEvent {
        id: event_id,
        aggregate_id: Uuid::new_v4(),
        payload: TestEvent { add: 1 },
        occurred_on: Utc::now(),
        sequence_number: 1,
        version: None,
    };

    bus.publish(&store_event).await;

    match consumer.try_next().await {
        Ok(delivery_opt) => {
            let delivery = delivery_opt.expect("error in consumer");
            delivery.ack(BasicAckOptions::default()).await.unwrap();
            let event = serde_json::from_slice::<StoreEvent<TestEvent>>(delivery.data.as_slice()).unwrap();
            assert_eq!(event.id, event_id);
        }
        Err(_) => {
            panic!("try next returned error");
        }
    }
}

async fn consumer(url: &str, exchange: &str, queue: &str, routing_key: &str) -> Consumer {
    let conn = Connection::connect(url, ConnectionProperties::default()).await.unwrap();

    let channel = conn.create_channel().await.unwrap();

    channel
        .queue_declare(queue, QueueDeclareOptions::default(), FieldTable::default())
        .await
        .unwrap();

    channel
        .queue_bind(
            queue,
            exchange,
            routing_key,
            QueueBindOptions::default(),
            FieldTable::default(),
        )
        .await
        .unwrap();

    channel
        .basic_consume(
            queue,
            "test_consumer",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await
        .unwrap()
}

#[allow(dead_code)]
pub fn random_letters() -> String {
    let mut rng = rand::thread_rng();
    let chars: String = (0..6)
        .map(|_| (b'a'..=b'z').choose(&mut rng).unwrap() as char)
        .collect();
    chars
}