feat(core): refactor core and publisher to integrate coherently

benchs/nats-publisher/src/utils/blocks.rs

@@ -2,7 +2,7 @@ use async_nats::jetstream::context::Publish;
 use fuel_core::combined_database::CombinedDatabase;
 use fuel_core_storage::transactional::AtomicView;
 use fuel_core_types::{blockchain::block::Block, fuel_types::BlockHeight};
-use fuel_streams_core::{blocks::BlocksSubject, nats::IntoSubject};
+use fuel_streams_core::{blocks::BlocksSubject, prelude::IntoSubject};
 use tokio::try_join;
 use tracing::info;
 

benchs/nats-publisher/src/utils/tx.rs

@@ -9,7 +9,7 @@ use fuel_core_types::{
 };
 use fuel_streams_core::{
     blocks::types::BlockHeight,
-    nats::IntoSubject,
+    prelude::IntoSubject,
     transactions::TransactionsSubject,
 };
 use tokio::try_join;
@@ -63,7 +63,7 @@ impl TxHelper {
         tx: &Transaction,
         index: usize,
     ) -> anyhow::Result<()> {
-        let subject = self.get_subject(tx, block, index);
+        let subject = &self.get_subject(tx, block, index);
         let payload = self
             .nats
             .data_parser()

crates/fuel-streams-core/Cargo.toml

@@ -13,6 +13,7 @@ version = "0.0.0"
 anyhow = { workspace = true }
 async-nats = { workspace = true }
 async-trait = { workspace = true }
+buildstructor = "0.5.4"
 bytes = { workspace = true }
 displaydoc = { workspace = true }
 dotenvy = { workspace = true }
@@ -21,6 +22,7 @@ fuel-core-client = { workspace = true }
 fuel-core-types = { workspace = true }
 fuel-data-parser = { path = "../fuel-data-parser" }
 fuel-streams-macros = { path = "../fuel-streams-macros" }
+futures = { workspace = true }
 futures-util = { workspace = true }
 pretty_assertions = { workspace = true, optional = true }
 rand = { workspace = true }

crates/fuel-streams-core/src/blocks/mod.rs

@@ -11,7 +11,4 @@ impl StreamEncoder for Block {}
 impl Streamable for Block {
     const NAME: &'static str = "blocks";
     const WILDCARD_LIST: &'static [&'static str] = &[BlocksSubject::WILDCARD];
-
-    type Builder = NatsStore<Self>;
-    type MainSubject = BlocksSubject;
 }

crates/fuel-streams-core/src/blocks/subjects.rs

@@ -10,16 +10,13 @@ pub struct BlocksSubject {
     pub height: Option<BlockHeight>,
 }
 
-impl From<Block> for BlocksSubject {
-    fn from(block: Block) -> Self {
-        BlocksSubject::new().with_height(Some(block.into()))
-    }
-}
-
 impl From<&Block> for BlocksSubject {
     fn from(block: &Block) -> Self {
         let block_height = *block.header().height();
-        BlocksSubject::new().with_height(Some(BlockHeight::from(block_height)))
+        BlocksSubject::new()
+        // TODO: Use correct block producer here
+            .with_producer(Some("0x".to_string()))
+            .with_height(Some(BlockHeight::from(block_height)))
     }
 }
 
@@ -32,7 +29,7 @@ mod test {
 
     #[test]
     fn block_subjects_all() {
-        assert_eq!(BlocksSubject::all(), "blocks.>")
+        assert_eq!(BlocksSubject::WILDCARD, "blocks.>")
     }
 
     #[test]
@@ -58,9 +55,10 @@ mod test {
 
     #[test]
     fn block_subjects_from_block() {
-        let mock_block = MockBlock::build(1);
-        let subject = BlocksSubject::from(mock_block.to_owned());
-        assert!(subject.producer.is_none());
-        assert_eq!(subject.height.unwrap(), mock_block.into());
+        let mock_block = &MockBlock::build(1);
+        let subject: BlocksSubject = mock_block.into();
+
+        assert!(subject.producer.is_some());
+        assert_eq!(subject.height.unwrap(), mock_block.clone().into());
     }
 }

crates/fuel-streams-core/src/lib.rs

@@ -1,12 +1,18 @@
 pub mod blocks;
 pub mod nats;
+mod stream;
 pub mod transactions;
 pub mod types;
 
+pub use stream::*;
+
 pub mod prelude {
+    pub use fuel_streams_macros::subject::*;
+
     pub use crate::{
         blocks::subjects::*,
         nats::*,
+        stream::*,
         transactions::subjects::*,
         types::*,
     };