Remove event filtering on client's side (#66)

Author: iamvukasin

backend/src/bin/client.rs

@@ -2,12 +2,8 @@ use alloy_primitives::B256;
 use clap::Parser;
 use execution_events_example::event_listener::EventName;
 use execution_events_example::serializable_event::SerializableExecEvent;
-use execution_events_example::{
-    event_filter::{ClientMessage, EventFilterSpec},
-    server::ServerMessage,
-};
-use execution_events_example::event_filter::load_restricted_filters;
-use futures_util::{SinkExt, StreamExt};
+use execution_events_example::server::ServerMessage;
+use futures_util::StreamExt;
 use std::collections::HashMap;
 use tokio_tungstenite::{connect_async, tungstenite::Message};
 
@@ -22,14 +18,6 @@ struct Cli {
     #[arg(short, long, default_value = "ws://127.0.0.1:3000")]
     url: String,
 
-    /// Filter events by type (comma-separated).
-    /// If not specified, all events are received.
-    #[arg(short, long, value_delimiter = ',')]
-    events: Option<Vec<String>>,
-
-    #[arg(long, default_value = "false")]
-    restricted: bool,
-
     #[arg(long, default_value = "false")]
     verbose_events: bool,
 
@@ -79,45 +67,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let (ws_stream, _) = connect_async(&cli.url).await?;
     info!("Connected!");
 
-    let (mut write, mut read) = ws_stream.split();
-
-    // Build event filter specs for subscription
-    let event_filters: Vec<EventFilterSpec> = if cli.restricted {
-        info!("Loading restricted filters");
-        let restricted_filters = load_restricted_filters();
-        restricted_filters.get_filter_specs()
-    } else {
-        let event_strings = cli.events.clone().unwrap_or_default();
-        if event_strings.is_empty() {
-            Vec::new()
-        } else {
-            let events: Vec<EventName> = event_strings
-                .iter()
-                .map(|s| {
-                    serde_json::from_value(serde_json::Value::String(s.clone()))
-                        .map_err(|_| format!("Invalid event name: {}", s))
-                })
-                .collect::<Result<Vec<_>, _>>()?;
-            events
-                .iter()
-                .map(|event_name| EventFilterSpec {
-                    event_name: *event_name,
-                    field_filters: Vec::new(),
-                })
-                .collect()
-        }
-    };
-
-    // Send subscription message
-    let subscribe_msg = ClientMessage::Subscribe { event_filters: event_filters.clone() };
-    let subscribe_json = serde_json::to_string(&subscribe_msg)?;
-    write.send(Message::Text(subscribe_json)).await?;
-
-    if event_filters.is_empty() {
-        info!("Subscribed to all events");
-    } else {
-        info!("Subscribed to {} event filters", event_filters.len());
-    }
+    let (_, mut read) = ws_stream.split();
 
     // Read messages from the server
     let mut events_per_sec_interval = tokio::time::interval(tokio::time::Duration::from_secs(1));

backend/src/lib/event_filter.rs

@@ -139,33 +139,6 @@ impl EventFilterSpec {
     }
 }
 
-/// Message sent by client to subscribe to specific event types
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(tag = "type")]
-pub enum ClientMessage {
-    /// Subscribe to events with filter specs.
-    /// Empty event_filters means subscribe to all events.
-    /// OR logic between specs: event passes if it matches ANY spec.
-    ///
-    /// Example: Subscribe to all BlockStart OR TxnLog with address=0xABC:
-    /// ```json
-    /// {
-    ///   "type": "subscribe",
-    ///   "event_filters": [
-    ///     { "event_name": "BlockStart" },
-    ///     { "event_name": "TxnLog", "field_filters": [
-    ///       { "field": "Address", "filter": { "values": ["0xABC"] } }
-    ///     ]}
-    ///   ]
-    /// }
-    /// ```
-    #[serde(rename = "subscribe")]
-    Subscribe {
-        #[serde(default)]
-        event_filters: Vec<EventFilterSpec>,
-    },
-}
-
 fn is_native_transfer(event: &SerializableEventData) -> bool {
     if let SerializableExecEvent::TxnCallFrame {
         value,

backend/src/lib/server.rs

@@ -14,7 +14,7 @@ use crate::event_filter::{is_restricted_mode, load_restricted_filters};
 use crate::event_listener::EventName;
 use crate::top_k_tracker::{AccessEntry, TopKTracker};
 
-use super::event_filter::{ClientMessage, EventFilter};
+use super::event_filter::EventFilter;
 use super::event_listener::EventData;
 use super::serializable_event::SerializableEventData;
 
@@ -67,61 +67,6 @@ impl TPSTracker {
     }
 }
 
-/// Wait for the client to send a subscription message, with a timeout.
-/// Returns the filter, or None if the client disconnects or times out.
-async fn wait_for_subscription(
-    ws_receiver: &mut SplitStream<WebSocketStream<TcpStream>>,
-    addr: SocketAddr,
-    restricted_filters: Option<EventFilter>,
-) -> Option<EventFilter> {
-    let timeout = tokio::time::Duration::from_secs(10);
-
-    match tokio::time::timeout(timeout, ws_receiver.next()).await {
-        Ok(Some(Ok(Message::Text(text)))) => match serde_json::from_str::<ClientMessage>(&text) {
-            Ok(ClientMessage::Subscribe { event_filters }) => {
-                let filter = EventFilter::new(event_filters.clone());
-                if let Some(restricted_filters) = restricted_filters {
-                    if filter != restricted_filters {
-                        warn!(
-                            "Client {} subscription does not match restricted filters, closing connection",
-                            addr
-                        );
-                        return None;
-                    }
-                }
-                if filter.accepts_all() {
-                    info!("Client {} subscribed to all events", addr);
-                } else {
-                    info!("Client {} subscribed with {} event filters", addr, event_filters.len());
-                }
-                Some(filter)
-            }
-            Err(e) => {
-                warn!(
-                    "Client {} sent invalid subscription: {} - {}",
-                    addr, e, text
-                );
-                None
-            }
-        },
-        Ok(Some(Ok(_))) => {
-            warn!("Client {} sent non-text message before subscribing", addr);
-            None
-        }
-        Ok(Some(Err(e))) => {
-            warn!("WebSocket error from {} before subscription: {}", addr, e);
-            None
-        }
-        Ok(None) => {
-            warn!("Client {} disconnected before subscribing", addr);
-            None
-        }
-        Err(_) => {
-            warn!("Client {} timed out waiting for subscription", addr);
-            None
-        }
-    }
-}
 
 // Helper function to process events into buffers
 fn process_event(
@@ -352,7 +297,7 @@ async fn handle_connection(
     stream: TcpStream,
     addr: SocketAddr,
     event_broadcast_receiver: broadcast::Receiver<EventDataOrMetrics>,
-    restricted_filters: Option<EventFilter>,
+    filter: EventFilter,
 ) {
     info!("New WebSocket connection from: {}", addr);
 
@@ -364,13 +309,7 @@ async fn handle_connection(
         }
     };
 
-    let (ws_sender, mut ws_receiver) = ws_stream.split();
-
-    // Wait for subscription message before streaming events
-    let filter = match wait_for_subscription(&mut ws_receiver, addr, restricted_filters).await {
-        Some(f) => f,
-        None => return,
-    };
+    let (ws_sender, ws_receiver) = ws_stream.split();
 
     // Spawn a task to receive events from the broadcast channel and send batches to this client
     let mut send_task = tokio::spawn(client_write_task(
@@ -381,7 +320,7 @@ async fn handle_connection(
     ));
 
     // Spawn a task to handle incoming messages from the client (mostly for ping/pong)
-    let mut recv_task = tokio::spawn(client_read_task(addr.clone(), ws_receiver));
+    let mut recv_task = tokio::spawn(client_read_task(addr, ws_receiver));
 
     // Wait for either task to finish
     tokio::select! {
@@ -420,20 +359,20 @@ pub async fn run_websocket_server(
     let listener = TcpListener::bind(&server_addr).await?;
     info!("WebSocket server listening on: {}", server_addr);
 
-    let restricted_filters: Option<EventFilter> = if is_restricted_mode() {
+    let filter: EventFilter = if is_restricted_mode() {
         info!("Running in restricted mode");
-        Some(load_restricted_filters())
+        load_restricted_filters()
     } else {
         info!("Running in unrestricted mode");
-        None
+        EventFilter::default()
     };
 
     // Accept incoming connections
     loop {
         match listener.accept().await {
-            Ok((stream, client_addr)) => {                        
+            Ok((stream, client_addr)) => {
                 let event_broadcast_receiver = event_broadcast_sender.subscribe();
-                tokio::spawn(handle_connection(stream, client_addr, event_broadcast_receiver, restricted_filters.clone()));
+                tokio::spawn(handle_connection(stream, client_addr, event_broadcast_receiver, filter.clone()));
             }
             Err(e) => {
                 error!("Error accepting connection: {}", e);