Add custom module for fil pubsub

Cargo.toml

@@ -161,6 +161,7 @@ reqwest = { version = "0.11.24", default-features = false, features = [
 ] } # use rustls instead of native (openSSL) tls to drop the number of build dependencies
 rlimit = "0.10.1"
 rs-car-ipfs = "0.3"
+rustc-hash = "1.1.0"
 rustyline = "13"
 scopeguard = "1.1.0"
 semver = "1.0"

src/chain/store/chain_store.rs

@@ -508,6 +508,7 @@ pub mod headchange_json {
     #[serde(tag = "type", content = "val")]
     pub enum HeadChangeJson {
         Apply(LotusJson<Tipset>),
+        Current(LotusJson<Tipset>),
     }
 
     impl From<HeadChange> for HeadChangeJson {

src/rpc/chain_api.rs

@@ -5,11 +5,12 @@
 use crate::blocks::{CachingBlockHeader, Tipset, TipsetKey};
 use crate::chain::index::ResolveNullTipset;
 use crate::chain::ChainStore;
+use crate::chain::HeadChange;
 use crate::cid_collections::CidHashSet;
 use crate::lotus_json::LotusJson;
 use crate::message::ChainMessage;
 use crate::rpc::error::JsonRpcError;
-use crate::rpc_api::data_types::{ApiMessage, ApiReceipt};
+use crate::rpc_api::data_types::{ApiHeadChange, ApiMessage, ApiReceipt};
 use crate::rpc_api::{
     chain_api::*,
     data_types::{ApiTipsetKey, BlockMessages, Data, RPCState},
@@ -28,6 +29,7 @@ use jsonrpsee::types::Params;
 use once_cell::sync::Lazy;
 use sha2::Sha256;
 use std::sync::Arc;
+use tokio::sync::broadcast::{self, Receiver as Subscriber};
 use tokio::sync::Mutex;
 
 pub async fn chain_get_message<DB: Blockstore>(
@@ -445,6 +447,33 @@ pub(crate) async fn chain_get_min_base_fee<DB: Blockstore>(
     Ok(min_base_fee.atto().to_string())
 }
 
+pub(crate) fn chain_notify<DB: Blockstore>(
+    _params: Params<'_>,
+    data: Arc<RPCState<DB>>,
+) -> Subscriber<ApiHeadChange> {
+    let (sender, receiver) = broadcast::channel(100);
+
+    // As soon as the channel is created, send the current tipset
+    let current = data.chain_store.heaviest_tipset();
+    let (change, headers) = ("current".into(), current.block_headers().clone().into());
+    let _ = sender.send(ApiHeadChange { change, headers });
+
+    let mut subscriber = data.chain_store.publisher().subscribe();
+
+    tokio::spawn(async move {
+        while let Ok(v) = subscriber.recv().await {
+            let (change, headers) = match v {
+                HeadChange::Apply(ts) => ("apply".into(), ts.block_headers().clone().into()),
+            };
+
+            if sender.send(ApiHeadChange { change, headers }).is_err() {
+                break;
+            }
+        }
+    });
+    receiver
+}
+
 fn load_api_messages_from_tipset(
     store: &impl Blockstore,
     tipset: &Tipset,