# Stream message (Carbon 2.0) > Feed account and transaction updates from your own code into a Carbon pipeline through a Tokio channel. `StreamMessageClient` wraps the receiving end of a Tokio `mpsc` channel. Anything you send into the channel as a `UnifiedMessage` reaches the pipeline as an account or transaction update. Use it to plug in a source Carbon doesn't ship (Kafka, a message queue, a file replay, your own Geyser plugin) or to drive a pipeline from tests, without implementing the [`Datasource` trait](https://carbon.sevenlabs.org/guides/custom-datasource.md) yourself. | | | | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------- | | Crate | `carbon-stream-message-datasource` 2.0.0 | | Install | `cargo add carbon-stream-message-datasource@2.0.0` | | Emits | `AccountUpdate`, `AccountDeletion`, `Transaction` | | Mode | Whatever your producer sends. Finishes when every sender is dropped. | | Provider | None. | | API docs | [docs.rs](https://docs.rs/carbon-stream-message-datasource/2.0.0/carbon_stream_message_datasource/) | | Source | [`datasources/stream-message-datasource`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/datasources/stream-message-datasource/src/lib.rs) | ## Quick example This program builds three Orca Whirlpool `FeeTier` accounts by hand, sends them through the channel, and lets the pipeline decode them. It runs offline. ```sh cargo new stream-message && cd stream-message cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-stream-message-datasource@2.0.0 cargo add solana-account@4.3.2 solana-pubkey@4.2.1 cargo add tokio@1 --features full cargo add env_logger@0.11 log@0.4 ``` `src/main.rs`: ```rust use { carbon_core::{ account::AccountProcessorInputType, datasource::AccountUpdate, error::CarbonResult, pipeline::Pipeline, processor::Processor, }, carbon_log_metrics::LogMetrics, carbon_orca_whirlpool_decoder::{ accounts::OrcaWhirlpoolAccount, OrcaWhirlpoolDecoder, PROGRAM_ID, }, carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage}, solana_account::Account, solana_pubkey::Pubkey, std::sync::Arc, }; #[tokio::main] async fn main() -> CarbonResult<()> { env_logger::init(); let (tx, rx) = tokio::sync::mpsc::channel::(1_000); let datasource = StreamMessageClient::new(rx); // Your producer: a Kafka consumer, a file replay, a test fixture... tokio::spawn(async move { for (slot, tick_spacing) in [(1_u64, 1_u16), (2, 64), (3, 128)] { let update = AccountUpdate { pubkey: Pubkey::new_unique(), account: fee_tier_account(tick_spacing), slot, transaction_signature: None, }; tx.send(UnifiedMessage::Account(update)).await.unwrap(); } // Dropping `tx` closes the stream; the pipeline exits once it drains. }); Pipeline::builder() .datasource(datasource) .metrics(Arc::new(LogMetrics::new())) .account(OrcaWhirlpoolDecoder, FeeTierLogger) .build()? .run() .await } /// Builds the raw bytes of an Orca `FeeTier` account: discriminator, config, tick spacing, fee rate. fn fee_tier_account(tick_spacing: u16) -> Account { let mut data = vec![56, 75, 159, 76, 142, 68, 190, 105]; data.extend_from_slice(Pubkey::new_unique().as_ref()); data.extend_from_slice(&tick_spacing.to_le_bytes()); data.extend_from_slice(&3000_u16.to_le_bytes()); Account { lamports: 1_200_000, data, owner: PROGRAM_ID, executable: false, rent_epoch: u64::MAX, } } struct FeeTierLogger; impl Processor> for FeeTierLogger { async fn process( &mut self, input: &AccountProcessorInputType<'_, OrcaWhirlpoolAccount>, ) -> CarbonResult<()> { if let OrcaWhirlpoolAccount::FeeTier(fee_tier) = &input.decoded_account.data { println!( "slot {} fee tier: tick_spacing={} default_fee_rate={}", input.metadata.slot, fee_tier.tick_spacing, fee_tier.default_fee_rate ); } Ok(()) } } ``` ```sh cargo run ``` Output: ```text slot 1 fee tier: tick_spacing=1 default_fee_rate=3000 slot 2 fee tier: tick_spacing=64 default_fee_rate=3000 slot 3 fee tier: tick_spacing=128 default_fee_rate=3000 ``` The program exits on its own after the third line. ## Configuration ### `StreamMessageClient::new` ```rust pub fn new(receiver: tokio::sync::mpsc::Receiver) -> Self ``` The channel's capacity is yours to choose; it buffers messages before they reach the pipeline's own channel. ### `UnifiedMessage` ```rust pub enum UnifiedMessage { Account(AccountUpdate), Transaction(Box), } ``` `AccountUpdate` and `TransactionUpdate` are the types from `carbon_core::datasource`; see the [updates reference](https://carbon.sevenlabs.org/reference/updates.md) for their fields. There is no variant for block details or explicit deletions. ### `handle_message_stream` ```rust pub async fn handle_message_stream( receiver: Receiver, cancellation_token: CancellationToken, sender: Sender<(Update, DatasourceId)>, id: DatasourceId, ) ``` The loop `StreamMessageClient` runs internally, exported for use in your own `Datasource` implementation. ## Behaviour - **Accounts.** Converted with `AccountUpdate::into_update()`: zero-lamport accounts become `AccountDeletion`, others `AccountUpdate`. - **Transactions.** Forwarded unchanged, so you set `signature`, `meta`, `slot`, `is_vote` and the rest yourself. Carbon does not filter failed transactions here. - **Ordering and backpressure.** Messages are forwarded in channel order with `send().await`; nothing is dropped. - **Finishing.** When every `Sender` for your channel is dropped, the loop logs `Receiver closed` and ends. A pipeline with only this datasource then shuts down. - **Single use.** The receiver is taken on the first `consume`. A second call logs `StreamMessageClient.consume() called more than once; receiver already taken` and does nothing. - **Metrics.** The names start with `agave_grpc_`: counters `agave_grpc_account_updates_received_total`, `agave_grpc_account_deletions_received_total`, `agave_grpc_transaction_updates_received_total`; histograms `agave_grpc_account_process_time_nanoseconds`, `agave_grpc_account_deletion_process_time_nanoseconds`, `agave_grpc_transaction_process_time_nanoseconds`. ## Troubleshooting ### The pipeline never exits A `Sender` clone is still alive somewhere, so the stream never closes. Drop every clone when your producer is done. ### The pipeline exits immediately Every sender was dropped before anything was sent, or the producer task panicked. Check that the producer holds `tx` for as long as it sends. ### Accounts aren't decoded The decoder checks `account.owner` and the account discriminator. Set `owner` to the program ID and put the raw on-chain bytes, discriminator included, in `data`. ### The README example doesn't compile The crate's README refers to `StreamMessageDatasource` and `pipeline.register`; neither exists. The type is `StreamMessageClient`, added with `Pipeline::builder().datasource(…)` as above. ## See also - [Custom datasource](https://carbon.sevenlabs.org/guides/custom-datasource.md) - [Testing](https://carbon.sevenlabs.org/guides/testing.md) and [test utilities](https://carbon.sevenlabs.org/reference/test-utils.md) - [Updates reference](https://carbon.sevenlabs.org/reference/updates.md) - [Datasources concept](https://carbon.sevenlabs.org/concepts/datasources.md) --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/datasources/stream-message · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt