flusso-queue-channel 0.10.1

In-process channel queue implementation for flusso.
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

use super::*;

#[tokio::test]
async fn publishes_and_consumes_in_order() {
    let (producer, mut consumer) = channel::<i32>(8);
    producer.publish(1).await.unwrap();
    producer.publish(2).await.unwrap();
    drop(producer); // close the queue so it drains then ends

    let (first, first_handle) = consumer.recv().await.unwrap().unwrap().into_parts();
    assert_eq!(first, 1);
    first_handle.ack().await.unwrap();

    let (item, handle) = consumer.recv().await.unwrap().unwrap().into_parts();
    assert_eq!(item, 2);
    handle.ack().await.unwrap();

    assert!(consumer.recv().await.unwrap().is_none());
}

#[tokio::test]
async fn publishing_to_a_closed_queue_errors() {
    let (producer, consumer) = channel::<i32>(1);
    drop(consumer);
    assert!(matches!(producer.publish(1).await, Err(QueueError::Closed)));
}