Repository: libsql/libsql
File: libsql-server/src/http/user/listen.rs (Rust)

Code before:
                })
            }),
    )
}

async fn listen_stream(
    store: NamespaceStore,
    namespace: NamespaceName,
    table: String,
    actions: Option<Vec<Action>>,
) -> impl Stream<Item = crate::Result<AggregatorEvent>> {
    async_stream::try_stream! {
        let _sub = Subscription::new(store.clone(), namespace.clone(), table.clone());
        let mut stream = store.subscribe(namespace, table);

        while let Some(item) = stream.next().await  {
            match item {
                Ok(msg) if filter_actions(&msg, &actions) => {
                    LISTEN_EVENTS_SENT.increment(1);
                    yield AggregatorEvent::Changes(msg);
                },
                Err(BroadcastStreamRecvError::Lagged(n)) => {
                    LISTEN_EVENTS_DROPPED.increment(n as u64);
                    yield AggregatorEvent::Error(LAGGED_MSG);
                },
                _ => {}
            }
        }
    }
}

fn filter_actions(msg: &BroadcastMsg, actions: &Option<Vec<Action>>) -> bool {
    actions.as_ref().map_or(true, |actions| {
        actions.iter().any(|action| {
            let count = match action {
                Action::DELETE => msg.delete,
                Action::INSERT => msg.insert,
                Action::UPDATE => msg.update,
                Action::UNKNOWN => msg.unknown,
            };
            count > 0
        })
    })
}

struct Subscription {
    store: NamespaceStore,
    namespace: NamespaceName,
    table: String,
}


Review comment (suggestion):
⛏️: this worried me a bit because you may end up ignoring any other kinds of errors that may come up as this code evolves 

Code after:
                })
            }),
    )
}

async fn listen_stream(
    store: NamespaceStore,
    namespace: NamespaceName,
    table: String,
    actions: Option<Vec<Action>>,
) -> impl Stream<Item = crate::Result<AggregatorEvent>> {
    async_stream::try_stream! {
        let _sub = Subscription::new(store.clone(), namespace.clone(), table.clone());
        let mut stream = store.subscribe(namespace.clone(), table.clone());

        while let Some(item) = stream.next().await {
            match item {
                Ok(msg) => if filter_actions(&msg, &actions) {
                    LISTEN_EVENTS_SENT.increment(1);
                    yield AggregatorEvent::Changes(msg);
                },
                Err(BroadcastStreamRecvError::Lagged(n)) => {
                    LISTEN_EVENTS_DROPPED.increment(n as u64);
                    yield AggregatorEvent::Error(LAGGED_MSG);
                },
            }
        }
    }
}

fn filter_actions(msg: &BroadcastMsg, actions: &Option<Vec<Action>>) -> bool {
    actions.as_ref().map_or(true, |actions| {
        actions.iter().any(|action| {
            let count = match action {
                Action::DELETE => msg.delete,
                Action::INSERT => msg.insert,
                Action::UPDATE => msg.update,
                Action::UNKNOWN => msg.unknown,
            };
            count > 0
        })
    })
}

struct Subscription {
    store: NamespaceStore,
    namespace: NamespaceName,
    table: String,
}

impl Subscription {