async fn consume_from_stream(
    channel: async_channel::Sender<ConsumerRecord>,
    mut stream: impl Stream<Item = Result<ConsumerRecord, ErrorCode>> + Unpin,
) -> Result<(), ()> {
    while let Ok(Some(record_raw)) = stream.try_next().await {
        channel.send(record_raw).await.expect("channel");
    }

    Ok(())
}
// This is for joining multiple topic support per process
This is for joining multiple topic support per process
This is for joining multiple topic support per process
async fn consume_from_stream(
    channel: async_channel::Sender<ConsumerRecord>,
    mut stream: impl Stream<Item = Result<ConsumerRecord, ErrorCode>> + Unpin,
) -> Result<(), ()> {
    while let Ok(Some(record_raw)) = stream.try_next().await {
        channel.send(record_raw).await.expect("channel");
    }
    Ok(())
}
async fn consume_from_stream ( channel : async_channel :: Sender < ConsumerRecord > , mut stream : impl Stream < Item = Result < ConsumerRecord , ErrorCode > > + Unpin , ) -> Result < ( ) , ( ) > { while let Ok ( Some ( record_raw ) ) = stream . try_next ( ) . await { channel . send ( record_raw ) . await . expect ( " " ) ; } Ok ( ( ) ) }
This is for joining multiple topic support per process
This is for joining multiple topic support per process
channel
async_channel::Sender<ConsumerRecord>
stream
impl Stream<Item = Result<ConsumerRecord, ErrorCode>> + Unpin
channel
async_channel::Sender<ConsumerRecord>
stream
impl Stream<Item = Result<ConsumerRecord, ErrorCode>> + Unpin