fix: end client rx upon disconnection error (#305)

Co-authored-by: Rui Fu <[email protected]>
streamnative · Mar 14, 2024 · a633e70 · a633e70
1 parent f51c9ea
commit a633e70
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/src/consumer/engine.rs b/src/consumer/engine.rs
@@ -103,6 +103,9 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                 let r = event_tx.send(mapper(Some(msg))).await;
                 if let Err(err) = r {
                     log::error!("Error sending event to channel - {err}");
+                    if err.is_disconnected() {
+                        break;
+                    }
                 }
             }
             let send_end_res = event_tx.send(mapper(None)).await;