feat(rpc): add ?topics= filtering to the events stream

crates/blockchain/src/events.rs

@@ -14,6 +14,7 @@ use ethlambda_types::ShortRoot;
 use ethlambda_types::checkpoint::Checkpoint;
 use ethlambda_types::primitives::H256;
 use serde::Serialize;
+use std::str::FromStr;
 use tokio::sync::broadcast;
 use tracing::warn;
 
@@ -44,6 +45,26 @@ impl Topic {
     }
 }
 
+/// Error returned by [`Topic::from_str`] for a name matching no topic.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("unknown topic: '{0}'")]
+pub struct UnknownTopic(String);
+
+impl FromStr for Topic {
+    type Err = UnknownTopic;
+
+    /// Exact inverse of [`Topic::as_str`].
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "head" => Ok(Topic::Head),
+            "block" => Ok(Topic::Block),
+            "justified_checkpoint" => Ok(Topic::JustifiedCheckpoint),
+            "finalized_checkpoint" => Ok(Topic::FinalizedCheckpoint),
+            other => Err(UnknownTopic(other.to_string())),
+        }
+    }
+}
+
 /// A consensus event published by the blockchain actor.
 ///
 /// Fields mirror the Ethereum beacon-API eventstream payloads so the SSE
@@ -276,6 +297,13 @@ mod tests {
         }
     }
 
+    const ALL_TOPICS: [Topic; 4] = [
+        Topic::Head,
+        Topic::Block,
+        Topic::JustifiedCheckpoint,
+        Topic::FinalizedCheckpoint,
+    ];
+
     #[tokio::test]
     async fn subscriber_receives_emitted_event() {
         let bus = EventBus::default();
@@ -289,6 +317,15 @@ mod tests {
         }
     }
 
+    #[test]
+    fn topic_from_str_inverts_as_str() {
+        for topic in ALL_TOPICS {
+            assert_eq!(topic.as_str().parse::<Topic>().unwrap(), topic);
+        }
+        let err = "bogus".parse::<Topic>().unwrap_err();
+        assert_eq!(err.to_string(), "unknown topic: 'bogus'");
+    }
+
     #[test]
     fn emit_without_subscribers_is_a_noop() {
         let bus = EventBus::default();

crates/blockchain/src/lib.rs

@@ -31,7 +31,7 @@ use crate::block_builder::ProposerConfig;
 use crate::events::ChainEventSnapshot;
 use crate::store::StoreError;
 
-pub use events::{ChainEvent, EventBus, Topic};
+pub use events::{ChainEvent, EventBus, Topic, UnknownTopic};
 
 pub mod aggregation;
 pub mod block_builder;

crates/net/rpc/Cargo.toml

@@ -26,8 +26,7 @@ serde_json.workspace = true
 hex.workspace = true
 tracing.workspace = true
 jemalloc_pprof.workspace = true
-tokio-stream = { version = "0.1", features = ["sync"] }
-futures-core = "0.3"
+futures-util = "0.3"
 
 [dev-dependencies]
 ethlambda-types.workspace = true