11use futures:: StreamExt ;
22use rabbitmq_stream_client:: error:: StreamCreateError ;
33use rabbitmq_stream_client:: types:: { ByteCapacity , OffsetSpecification , ResponseCode } ;
4- use std:: io:: stdin;
54use std:: sync:: atomic:: { AtomicI64 , Ordering } ;
65use std:: sync:: Arc ;
6+ use tokio:: sync:: Notify ;
77use tokio:: task;
88
99#[ tokio:: main]
@@ -14,6 +14,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
1414let received_messages =Arc :: new ( AtomicI64 :: new ( -1 ) ) ;
1515let first_offset =Arc :: new ( AtomicI64 :: new ( -1 ) ) ;
1616let last_offset =Arc :: new ( AtomicI64 :: new ( -1 ) ) ;
17+ let notify_on_close =Arc :: new ( Notify :: new ( ) ) ;
1718let create_response = environment
1819. stream_creator ( )
1920. max_length ( ByteCapacity :: GB ( 2 ) )
@@ -41,8 +42,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
4142. await
4243. unwrap ( ) ;
4344
44- println ! ( "Starting consuming" ) ;
45- println ! ( "Press any key to close the consumer" ) ;
45+ println ! ( "Started consuming" ) ;
4646
4747let mut stored_offset: u64 = consumer. query_offset ( ) . await . unwrap_or_else ( |_|0 ) ;
4848
@@ -59,13 +59,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
5959
6060let first_cloned_offset = first_offset. clone ( ) ;
6161let last_cloned_offset = last_offset. clone ( ) ;
62+ let notify_on_close_cloned = notify_on_close. clone ( ) ;
6263
6364 task:: spawn ( async move {
6465while let Some ( delivery) = consumer. next ( ) . await {
6566let d = delivery. unwrap ( ) ;
6667
6768if first_offset. load ( Ordering :: Relaxed ) == -1 {
68- println ! ( "consuming first message" ) ;
69+ println ! ( "First message received " ) ;
6970 _ = first_offset. compare_exchange (
7071 first_offset. load ( Ordering :: Relaxed ) ,
7172 d. offset ( ) as i64 ,
@@ -85,13 +86,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
8586 last_offset. store ( d. offset ( ) as i64 , Ordering :: Relaxed ) ;
8687let handle = consumer. handle ( ) ;
8788 _ = handle. close ( ) . await ;
89+ notify_on_close_cloned. notify_one ( ) ;
8890
8991}
9092}
9193}
9294} ) ;
9395
94- _ = stdin ( ) . read_line ( & mut "" . to_string ( ) ) ;
96+ notify_on_close . notified ( ) . await ;
9597
9698if first_cloned_offset. load ( Ordering :: Relaxed ) != -1 {
9799println ! (