iconoclast 0.3.1

Reusable code for Rust-based business μServices
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

use crate::kafka::{MessageHandler, consumer};
use consumer::Consumer;
use rdkafka::message::BorrowedMessage;
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::{ClientConfig, Message};
use std::collections::HashMap;
use std::convert::Infallible;
use tokio::sync::oneshot;

async fn publish(brokers: String, topic: &str, key: &str, payload: &str) {
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .create()
        .unwrap();

    let record = FutureRecord::to(topic).payload(payload).key(key);
    producer.send(record, None).await.unwrap();
}

#[faux::create]
struct TestHandler {}

#[faux::methods]
impl MessageHandler<Infallible> for TestHandler {
    type Message = Pl;

    fn topics() -> &'static [&'static str] {
        &["hello"]
    }

    async fn handle(&self, _payload: Pl) -> Result<(), Infallible> {
        unimplemented!("mock")
    }
}

#[derive(Debug, PartialEq, Eq)]
struct Pl(String);

impl TryFrom<&BorrowedMessage<'_>> for Pl {
    type Error = Infallible;

    fn try_from(bm: &BorrowedMessage) -> Result<Self, Self::Error> {
        let payload = bm.payload();
        let payload = payload.unwrap();
        let message = String::from_utf8(Vec::from(payload));

        Ok(Self(message.unwrap()))
    }
}

/// test that a message sent through kafka ends up in call to the (mock-) Service
#[tokio::test]
async fn smoketest() {
    // set up a mock that sends the value (once), as we need to wait for the value
    let (tx, rx) = oneshot::channel::<Pl>();
    // let tx = RefCell::new(Some(tx));
    let mut service = TestHandler::faux();
    faux::when!(service.handle).once().then(move |m| {
        tx.send(m).unwrap();
        Ok(())
    });

    // the test never exits. mem::drop() on the cluster doesn't help, this… seems to help somehow :/
    let cluster = Box::leak(Box::new(rdkafka::mocking::MockCluster::new(3).unwrap()));
    cluster.create_topic("hello", 12, 3).unwrap();

    let config = super::Config {
        env_properties: vec![
            ("bootstrap.servers".to_string(), cluster.bootstrap_servers()),
            ("group.id".to_string(), "smoketest".to_string()),
            ("auto.offset.reset".to_string(), "earliest".to_string()),
            ("enable.auto.commit".to_string(), "false".to_string()),
        ],
        properties: HashMap::new(),
    };

    let consumer = Consumer::new(&config, service).unwrap();

    let task = tokio::task::spawn(async move { consumer.start().await });

    publish(cluster.bootstrap_servers(), "hello", "1", "Ferris").await;

    let actual = rx.await.unwrap();
    assert_eq!(Pl("Ferris".to_string()), actual);
    task.abort();
}