Re-written FirehoseBlockStream using async-stream (#3078)

The implementation of the never ending Firehose stream of blocks have been re-implemented using `async-stream` library. This will makes the implementation much more readable and maintainable in the future. This should enable re-using `TriggersAdapter::triggers_in_block` directly instead of having to defer to `firehose_triggers_in_block` (subsequent PR to come).

This also removes the `notify_block_consumed` callback on the `FirehoseBlockStream` and the latest active cursor to reconnect to is now saved inside the FirehoseBlockStream directly. This will make it easier to implement the `BufferedBlockStream` in the future.

*Caveats* Formatting of code inside `try_stream!` macro doesn't work, unclear if it's a general limitation in macros in general or this very specific macro.

Co-authored-by: Matthieu Vachon <[email protected]>

Author: Matthieu Vachon

Cargo.lock

@@ -671,14 +671,6 @@ where
                         };
                     }
 
-                    // Notify the BlockStream implementation that a block was succesfully consumed
-                    // and that its internal cursoring mechanism can be saved to memory.
-                    //
-                    // The first `get_mut` is to get the inner `Stream` out of `Cancelable` which
-                    // returns a `TryStreamExt::MapErr` struct and the second `get_mut` is to get
-                    // out the actual `dyn BlockStream` trait on which we can call our method.
-                    block_stream.get_mut().get_mut().notify_block_consumed();
-
                     if needs_restart {
                         // Cancel the stream for real
                         ctx.state

core/src/subgraph/instance_manager.rs

@@ -6,6 +6,7 @@ edition = "2018"
 [dependencies]
 anyhow = "1.0"
 async-trait = "0.1.50"
+async-stream = "0.3"
 atomic_refcell = "0.1.8"
 bigdecimal = { version = "0.1.0", features = ["serde"] }
 bytes = "1.0.1"

graph/Cargo.toml

@@ -11,7 +11,6 @@ use crate::{prelude::*, prometheus::labels};
 pub trait BlockStream<C: Blockchain>:
     Stream<Item = Result<BlockStreamEvent<C>, Error>> + Unpin
 {
-    fn notify_block_consumed(&mut self) {}
 }
 
 pub type FirehoseCursor = Option<String>;