graph, store: Remove unused facilities for filtering StoreEventStreams

Author: lutter

graph/src/components/store/mod.rs

@@ -18,7 +18,7 @@ use strum_macros::Display;
 pub use traits::*;
 pub use write::Batch;
 
-use futures01::{Async, Stream};
+use futures01::Stream;
 use serde::{Deserialize, Serialize};
 use std::collections::btree_map::Entry;
 use std::collections::{BTreeMap, BTreeSet, HashSet};
@@ -633,37 +633,8 @@ impl PartialEq for StoreEvent {
     }
 }
 
-/// A `StoreEventStream` produces the `StoreEvents`. Various filters can be applied
-/// to it to reduce which and how many events are delivered by the stream.
-pub struct StoreEventStream<S> {
-    source: S,
-}
-
 /// A boxed `StoreEventStream`
-pub type StoreEventStreamBox =
-    StoreEventStream<Box<dyn Stream<Item = Arc<StoreEvent>, Error = ()> + Send>>;
-
-impl<S> Stream for StoreEventStream<S>
-where
-    S: Stream<Item = Arc<StoreEvent>, Error = ()> + Send,
-{
-    type Item = Arc<StoreEvent>;
-    type Error = ();
-
-    fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Self::Error> {
-        self.source.poll()
-    }
-}
-
-impl<S> StoreEventStream<S>
-where
-    S: Stream<Item = Arc<StoreEvent>, Error = ()> + Send + 'static,
-{
-    // Create a new `StoreEventStream` from another such stream
-    pub fn new(source: S) -> Self {
-        StoreEventStream { source }
-    }
-}
+pub type StoreEventStreamBox = Box<dyn Stream<Item = Arc<StoreEvent>, Error = ()> + Send>;
 
 /// An entity operation that can be transacted into the store.
 #[derive(Clone, Debug, PartialEq)]

graph/src/lib.rs

@@ -131,8 +131,8 @@ pub mod prelude {
         EntityCollection, EntityFilter, EntityLink, EntityOperation, EntityOrder,
         EntityOrderByChild, EntityOrderByChildInfo, EntityQuery, EntityRange, EntityWindow,
         EthereumCallCache, ParentLink, PartialBlockPtr, PoolWaitStats, QueryStore,
-        QueryStoreManager, StoreError, StoreEvent, StoreEventStream, StoreEventStreamBox,
-        SubgraphStore, UnfailOutcome, WindowAttribute, BLOCK_NUMBER_MAX,
+        QueryStoreManager, StoreError, StoreEvent, StoreEventStreamBox, SubgraphStore,
+        UnfailOutcome, WindowAttribute, BLOCK_NUMBER_MAX,
     };
     pub use crate::components::subgraph::{
         BlockState, HostMetrics, InstanceDSTemplateInfo, RuntimeHost, RuntimeHostBuilder,

store/postgres/src/store_events.rs

@@ -221,6 +221,6 @@ impl SubscriptionManagerTrait for SubscriptionManager {
         self.subscriptions.write().unwrap().insert(id, sender);
 
         // Return the subscription ID and entity change stream
-        StoreEventStream::new(Box::new(ReceiverStream::new(receiver).map(Ok).compat()))
+        Box::new(ReceiverStream::new(receiver).map(Ok).compat())
     }
 }