Traits

Every trait you implement or call in Carbon 2.0.0, with required methods, associated types and a compiled implementation example.

Carbon is built from a few traits: datasources produce updates, decoders turn raw bytes into types, processors handle the result, filters gate pipes, and exporters publish metrics. This page lists each trait in carbon-core 2.0.0 with its exact definition and an implementation that compiles against the published crates.

Examples on this page assume these dependencies:

Cargo.toml

[dependencies]
async-trait = "0.1"
borsh = { version = "1.5.7", features = ["derive"] }
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["serde"] }
carbon-token-program-decoder = { version = "2.0.0", features = ["serde"] }
serde = { version = "1", features = ["derive"] }
solana-account = "4.3.2"
solana-instruction = "3.4.1"
solana-pubkey = "4.2.1"
tokio = { version = "1", features = ["full"] }
tokio-util = "0.7"

Trait	Module	You implement it to
`Datasource`	`carbon_core::datasource`	Feed updates from a new source
`Processor`	`carbon_core::processor`	Handle decoded data
`AccountDecoder`	`carbon_core::account`	Decode accounts of a program without a generated decoder
`InstructionDecoder`	`carbon_core::instruction`	Decode instructions of a program without a generated decoder
`CarbonDeserialize`	`carbon_core::deserialize`	Decode log events with `decode_log_events`
`ArrangeAccounts`	`carbon_core::deserialize`	Map an instruction's account list to named fields
`InstructionDecoderCollection`	`carbon_core::collection`	Use transaction pipes over one or more programs
`Filter`	`carbon_core::filter`	Gate which updates reach a pipe
`MetricsExporter`	`carbon_core::metrics`	Send metrics to a backend
`Metric`	`carbon_core::metrics`	Read a metric's name and help text

`Datasource`#

Produces updates and sends them into the pipeline.

Rust

#[async_trait]
pub trait Datasource: Send + Sync {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: tokio::sync::mpsc::Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()>;

    fn update_types(&self) -> Vec<UpdateType>;
}

Details. Datasource uses #[async_trait] because the pipeline stores datasources as Arc<dyn Datasource>; put #[async_trait::async_trait] on your impl too.

Method	Required	Description
`consume`	Yes	Called once by `Pipeline::run`, in its own Tokio task. Send each update as `(update, id.clone())` on `sender`. Return when `cancellation_token` is cancelled. Returning (with `Ok` or `Err`) ends this datasource; the pipeline stops when all datasources have returned and the channel is empty. An `Err` is logged as `error consuming datasource: …` and does not stop other datasources.
`update_types`	Yes	The `UpdateType`s this datasource emits. Not used by the pipeline in 2.0.0, but return an accurate list for readers and future versions.

Use sender.send(...).await to wait when the pipeline is behind (backpressure), or try_send to drop updates instead. Build account updates with AccountUpdate::into_update so zero-lamport accounts become deletions. &self is shared, so keep mutable state inside consume or behind a lock.

Example.

src/ticker.rs

use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
    },
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::time::Duration,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Emits one synthetic account update per second until cancelled.
pub struct Ticker {
    pub owner: Pubkey,
}

#[async_trait]
impl Datasource for Ticker {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let mut slot: u64 = 0;
        loop {
            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
            }
            slot += 1;

            let mut data = b"counter!".to_vec();
            data.extend_from_slice(&slot.to_le_bytes());

            let update = AccountUpdate {
                pubkey: Pubkey::new_unique(),
                account: Account {
                    lamports: 1_000_000,
                    data,
                    owner: self.owner,
                    executable: false,
                    rent_epoch: 0,
                },
                slot,
                transaction_signature: None,
            }
            .into_update();

            sender
                .send((update, id.clone()))
                .await
                .map_err(|e| Error::FailedToConsumeDatasource(e.to_string()))?;
        }
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::AccountUpdate]
    }
}

See also. Custom datasource, Updates, docs.rs.

`Processor`#

Handles one input value at a time.

Rust

pub trait Processor<T>
where
    T: Sync,
{
    fn process(&mut self, data: &T) -> impl Future<Output = CarbonResult<()>> + Send;
}

Details. Implement it with a native async fn process(&mut self, data: &T) -> CarbonResult<()>; no #[async_trait] is needed. T is the pipe's input type, listed in Processor inputs. For input types with a lifetime, write impl Processor<InstructionProcessorInputType<'_, MyInstruction>> for MyProcessor; the builder requires the impl for every lifetime, which '_ provides.

The pipeline owns the processor and calls process sequentially, so &mut self state needs no locking. The returned future must be Send: don't hold a std::sync::MutexGuard or Rc across an .await. Returning Err logs the error, increments carbon_updates_failed_total, skips the remaining pipes for that update and continues with the next update.

Example.

src/deletions.rs

use carbon_core::{datasource::AccountDeletion, error::CarbonResult, processor::Processor};

#[derive(Default)]
pub struct DeletionCounter {
    pub seen: u64,
}

impl Processor<AccountDeletion> for DeletionCounter {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        self.seen += 1;
        println!("closed {} at slot {} ({} so far)", deletion.pubkey, deletion.slot, self.seen);
        Ok(())
    }
}

See also. Processors, Processor inputs, docs.rs.

`AccountDecoder`#

Turns a raw account into a typed value.

Rust

pub trait AccountDecoder<'a> {
    type AccountType;

    fn decode_account(
        &self,
        account: &'a solana_account::Account,
    ) -> Option<DecodedAccount<Self::AccountType>>;
}

Details.

Item	Description
`type AccountType`	What your processor receives in `decoded_account.data`. Must be `Send + Sync + 'static` to register.
`decode_account`	Return `None` for accounts you don't handle; the pipe then skips the update. Check `account.owner` first.

Generated decoders implement this trait on their decoder struct (MeteoraDlmmDecoder, TokenProgramDecoder, …): they check the owner, then try each account type's discriminator in turn. Implement it by hand for programs without an IDL. Register the decoder with account, which requires the impl for every lifetime (impl<'a> AccountDecoder<'a> for …).

Example.

src/counter.rs

use {
    carbon_core::account::{AccountDecoder, DecodedAccount},
    solana_account::Account,
    solana_pubkey::Pubkey,
};

pub const COUNTER_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("Counter111111111111111111111111111111111111");

#[derive(Debug, Clone)]
pub struct CounterAccount {
    pub count: u64,
}

pub struct CounterDecoder;

impl<'a> AccountDecoder<'a> for CounterDecoder {
    type AccountType = CounterAccount;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<CounterAccount>> {
        if account.owner != COUNTER_PROGRAM_ID {
            return None;
        }
        let payload = account.data.strip_prefix(b"counter!")?;
        let count = u64::from_le_bytes(payload.get(..8)?.try_into().ok()?);

        Some(DecodedAccount {
            lamports: account.lamports,
            data: CounterAccount { count },
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

See also. DecodedAccount, Custom decoder, docs.rs.

`InstructionDecoder`#

Turns a raw instruction into a typed value.

Rust

pub trait InstructionDecoder<'a> {
    type InstructionType;

    fn decode_instruction(
        &self,
        instruction: &'a solana_instruction::Instruction,
    ) -> Option<Self::InstructionType>;
}

Details.

Item	Description
`type InstructionType`	What your processor receives in `decoded_instruction`. Must be `Send + Sync + 'static` to register.
`decode_instruction`	Called for every instruction of every transaction, top-level and inner. Return `None` quickly for other programs by checking `instruction.program_id` first.

The instruction has resolved AccountMetas, so you can read is_signer and is_writable. Generated decoders return an enum with one struct variant per instruction ({ program_id, data, accounts }) built with try_decode_instructions!.

Example.

src/memo.rs

use {
    carbon_core::instruction::InstructionDecoder,
    solana_instruction::Instruction,
    solana_pubkey::Pubkey,
};

pub const MEMO_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr");

#[derive(Debug, Clone)]
pub struct Memo {
    pub text: String,
    pub signers: Vec<Pubkey>,
}

pub struct MemoDecoder;

impl<'a> InstructionDecoder<'a> for MemoDecoder {
    type InstructionType = Memo;

    fn decode_instruction(&self, instruction: &'a Instruction) -> Option<Memo> {
        if instruction.program_id != MEMO_PROGRAM_ID {
            return None;
        }
        Some(Memo {
            text: String::from_utf8(instruction.data.clone()).ok()?,
            signers: instruction
                .accounts
                .iter()
                .filter(|meta| meta.is_signer)
                .map(|meta| meta.pubkey)
                .collect(),
        })
    }
}

See also. Instructions, Custom decoder, docs.rs.

`CarbonDeserialize`#

Discriminator-prefixed Borsh deserialization.

Rust

pub trait CarbonDeserialize
where
    Self: Sized + borsh::BorshDeserialize,
{
    const DISCRIMINATOR: &'static [u8];

    fn deserialize(data: &[u8]) -> Option<Self>;
}

Details.

Item	Description
`const DISCRIMINATOR`	The prefix that identifies this type, usually 8 bytes for Anchor. Not read by Carbon itself; use it in `deserialize`.
`deserialize`	Check and strip the discriminator, then Borsh-decode the rest. Return `None` on mismatch.

In 2.0.0 the only consumer is InstructionMetadata::decode_log_events. Generated decoders do not implement this trait: their account, instruction and event types have an inherent decode(&[u8]) -> Option<Self> method instead. To decode a generated event type from logs, wrap it or implement the trait on your own type. carbon_core::deserialize::extract_discriminator(length, data) splits data into (discriminator, rest) and returns None when data is shorter than length.

Example.

src/events.rs

use {
    borsh::BorshDeserialize,
    carbon_core::{
        deserialize::CarbonDeserialize, error::CarbonResult,
        instruction::InstructionProcessorInputType, processor::Processor,
    },
    carbon_meteora_dlmm_decoder::instructions::MeteoraDlmmInstruction,
    solana_pubkey::Pubkey,
};

/// An Anchor `emit!` event: 8-byte discriminator, then the Borsh body.
#[derive(Debug, BorshDeserialize)]
pub struct MyEvent {
    pub pool: Pubkey,
    pub amount: u64,
}

impl CarbonDeserialize for MyEvent {
    const DISCRIMINATOR: &'static [u8] = &[0x1d, 0x2c, 0x3b, 0x4a, 0x59, 0x68, 0x77, 0x86];

    fn deserialize(data: &[u8]) -> Option<Self> {
        let body = data.strip_prefix(Self::DISCRIMINATOR)?;
        borsh::from_slice(body).ok()
    }
}

pub struct EventLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for EventLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        for event in input.metadata.decode_log_events::<MyEvent>() {
            println!("{} moved {}", event.pool, event.amount);
        }
        Ok(())
    }
}

See also. Instructions, docs.rs.

`ArrangeAccounts`#

Maps an instruction's positional accounts to a struct.

Rust

pub trait ArrangeAccounts {
    type ArrangedAccounts: Clone + Send + Sync + std::fmt::Debug;

    fn arrange_accounts(
        accounts: &[solana_instruction::AccountMeta],
    ) -> Option<Self::ArrangedAccounts>;
}

Details. An associated function (no self), implemented on the instruction's data type. Return None when required accounts are missing; try_decode_instructions! then rejects the instruction. Generated decoders implement it for every instruction, treat optional accounts as Option<Pubkey>, and collect extra accounts into remaining: Vec<AccountMeta>. carbon_core::account_utils::next_account(&mut iter) returns the next account's Pubkey, or None at the end.

Example.

src/transfer.rs

use {
    carbon_core::{account_utils::next_account, deserialize::ArrangeAccounts},
    solana_instruction::AccountMeta,
    solana_pubkey::Pubkey,
};

pub struct Transfer;

#[derive(Debug, Clone)]
pub struct TransferAccounts {
    pub source: Pubkey,
    pub destination: Pubkey,
    pub authority: Pubkey,
    pub remaining: Vec<AccountMeta>,
}

impl ArrangeAccounts for Transfer {
    type ArrangedAccounts = TransferAccounts;

    fn arrange_accounts(accounts: &[AccountMeta]) -> Option<TransferAccounts> {
        let mut iter = accounts.iter();
        let source = next_account(&mut iter)?;
        let destination = next_account(&mut iter)?;
        let authority = next_account(&mut iter)?;

        Some(TransferAccounts {
            source,
            destination,
            authority,
            remaining: iter.as_slice().to_vec(),
        })
    }
}

See also. try_decode_instructions!, docs.rs.

`InstructionDecoderCollection`#

A single enum over the instructions of several programs, used by transaction pipes.

Rust

pub trait InstructionDecoderCollection:
    Clone + std::fmt::Debug + Send + Sync + Eq + std::hash::Hash + Serialize + 'static
{
    type InstructionType: Clone + std::fmt::Debug + PartialEq + Eq + Send + Sync + 'static;

    fn parse_instruction(instruction: &solana_instruction::Instruction) -> Option<Self>;

    fn get_type(&self) -> Self::InstructionType;
}

Details.

Item	Description
`type InstructionType`	A tag enum identifying the variant. Carbon does not read it in 2.0.0.
`parse_instruction`	Associated function (no decoder instance). Return the variant for the instruction's program, or `None`. Called on every instruction of every transaction.
`get_type`	Return the tag for `self`.

The supertraits include Eq, Hash and serde::Serialize. Generated decoder enums derive only Debug, Clone and PartialEq (plus Serialize/Deserialize with the serde feature), so wrap them in your own enum and implement Eq and Hash by hand, as below. The instruction_decoder_collection! macro generates this trait, but only for instruction enums that derive Hash, Eq and InstructionType, which generated decoders don't.

Example.

src/collection.rs

use {
    carbon_core::{collection::InstructionDecoderCollection, instruction::InstructionDecoder},
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder},
    carbon_token_program_decoder::{instructions::TokenProgramInstruction, TokenProgramDecoder},
    solana_instruction::Instruction,
    std::hash::{Hash, Hasher},
};

#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub enum AllInstructions {
    MeteoraDlmm(MeteoraDlmmInstruction),
    Token(TokenProgramInstruction),
}

// Generated instruction enums derive `PartialEq` only, so provide `Eq` and
// `Hash` by hand. Hashing the variant alone is consistent with `PartialEq`.
impl Eq for AllInstructions {}

impl Hash for AllInstructions {
    fn hash<H: Hasher>(&self, state: &mut H) {
        std::mem::discriminant(self).hash(state);
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AllInstructionTypes {
    MeteoraDlmm,
    Token,
}

impl InstructionDecoderCollection for AllInstructions {
    type InstructionType = AllInstructionTypes;

    fn parse_instruction(instruction: &Instruction) -> Option<Self> {
        if let Some(ix) = MeteoraDlmmDecoder.decode_instruction(instruction) {
            return Some(Self::MeteoraDlmm(ix));
        }
        TokenProgramDecoder
            .decode_instruction(instruction)
            .map(Self::Token)
    }

    fn get_type(&self) -> AllInstructionTypes {
        match self {
            Self::MeteoraDlmm(_) => AllInstructionTypes::MeteoraDlmm,
            Self::Token(_) => AllInstructionTypes::Token,
        }
    }
}

`Filter`#

Decides whether an update reaches a pipe.

Rust

pub trait Filter: Send + Sync {
    fn filter_account(&self, _context: &FilterContext, _account_metadata: &AccountMetadata, _account: &solana_account::Account) -> FilterResult { FilterResult::Accept }
    fn filter_instruction(&self, _context: &FilterContext, _nested_instruction: &NestedInstruction) -> FilterResult { FilterResult::Accept }
    fn filter_transaction(&self, _context: &FilterContext, _transaction_metadata: &TransactionMetadata, _nested_instructions: &NestedInstructions) -> FilterResult { FilterResult::Accept }
    fn filter_account_deletion(&self, _context: &FilterContext, _account_deletion: &AccountDeletion) -> FilterResult { FilterResult::Accept }
    fn filter_block_details(&self, _context: &FilterContext, _block_details: &BlockDetails) -> FilterResult { FilterResult::Accept }
}

Details. Every method has a default that returns Accept, so implement only the hooks for the pipe kinds you filter. Hooks take &self: use atomics or a lock for state, as DeduplicationFilter does. See the filters reference for when each hook runs.

Example.

src/filters.rs

use {
    carbon_core::{
        filter::{Filter, FilterContext, FilterResult},
        instruction::{NestedInstruction, NestedInstructions},
        transaction::TransactionMetadata,
    },
};

/// Drops failed transactions and every instruction inside them.
pub struct SuccessfulOnly;

impl Filter for SuccessfulOnly {
    fn filter_instruction(
        &self,
        _context: &FilterContext,
        nested_instruction: &NestedInstruction,
    ) -> FilterResult {
        if nested_instruction.metadata.transaction_metadata.meta.status.is_ok() {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }

    fn filter_transaction(
        &self,
        _context: &FilterContext,
        transaction_metadata: &TransactionMetadata,
        _nested_instructions: &NestedInstructions,
    ) -> FilterResult {
        if transaction_metadata.meta.status.is_ok() {
            FilterResult::Accept
        } else {
            FilterResult::Reject
        }
    }
}

See also. Filters reference, Filters, docs.rs.

`MetricsExporter`#

Publishes metric snapshots to a backend.

Rust

pub trait MetricsExporter: Send + Sync {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> {
        let _ = self;
        Ok(())
    }

    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()>;

    fn shutdown(&self) -> CarbonResult<()> {
        Ok(())
    }
}

Details.

Method	Required	When the pipeline calls it
`initialize`	No	Once, at the start of `run`, with an `Arc` of the exporter. Start periodic work here (for example `tokio::spawn` a loop that calls `self.export(&MetricsRegistry::global().snapshot())`).
`export`	Yes	Once, when the pipeline stops, with a final snapshot. Not called while the pipeline runs.
`shutdown`	No	Once, after the final `export`.

An Err from any of these makes Pipeline::run return that error. Register with PipelineBuilder::metrics.

Example.

src/exporter.rs

use {
    carbon_core::{
        error::CarbonResult,
        metrics::{MetricsExporter, MetricsRegistry, MetricsSnapshot},
    },
    std::{sync::Arc, time::Duration},
};

/// Prints every counter every 10 seconds, and once more at shutdown.
pub struct StdoutExporter;

impl MetricsExporter for StdoutExporter {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> {
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            loop {
                interval.tick().await;
                let _ = self.export(&MetricsRegistry::global().snapshot());
            }
        });
        Ok(())
    }

    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()> {
        for (name, _help, value) in &snapshot.counters {
            println!("{name} {value}");
        }
        Ok(())
    }
}

See also. Metrics reference, docs.rs.

`Metric`#

Name and help text shared by Counter, Gauge and Histogram.

Rust

pub trait Metric: Send + Sync {
    fn name(&self) -> &'static str;
    fn help(&self) -> &'static str {
        ""
    }
}

Details. Implemented by the three metric types in carbon_core::metrics. MetricsRegistry only stores &'static Counter, &'static Gauge and &'static Histogram, so implementing Metric on your own type does not make it exportable. To add a metric, declare a static Counter, Gauge or Histogram and register it.

Example.

src/metrics.rs

use carbon_core::metrics::{Counter, Metric, MetricsRegistry};

static SWAPS_SEEN: Counter = Counter::new("my_indexer_swaps_seen_total", "Swaps decoded");

pub fn register() {
    MetricsRegistry::global().register_counter(&SWAPS_SEEN);
    assert_eq!(SWAPS_SEEN.name(), "my_indexer_swaps_seen_total");
}

See also. Metrics reference, docs.rs.

Complete example#

This program implements a datasource and an account decoder, runs them for 3.5 seconds and needs no network.

src/main.rs

use {
    async_trait::async_trait,
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::{AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::{CarbonResult, Error},
        pipeline::Pipeline,
        processor::Processor,
    },
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::time::Duration,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

const COUNTER_PROGRAM_ID: Pubkey =
    Pubkey::from_str_const("Counter111111111111111111111111111111111111");

/// Emits one synthetic account update per second until cancelled.
struct Ticker;

#[async_trait]
impl Datasource for Ticker {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        let mut slot: u64 = 0;
        loop {
            tokio::select! {
                _ = cancellation_token.cancelled() => return Ok(()),
                _ = tokio::time::sleep(Duration::from_secs(1)) => {}
            }
            slot += 1;

            let mut data = b"counter!".to_vec();
            data.extend_from_slice(&slot.to_le_bytes());

            let update = AccountUpdate {
                pubkey: Pubkey::new_unique(),
                account: Account {
                    lamports: 1_000_000,
                    data,
                    owner: COUNTER_PROGRAM_ID,
                    executable: false,
                    rent_epoch: 0,
                },
                slot,
                transaction_signature: None,
            }
            .into_update();

            sender
                .send((update, id.clone()))
                .await
                .map_err(|e| Error::FailedToConsumeDatasource(e.to_string()))?;
        }
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::AccountUpdate]
    }
}

#[derive(Debug, Clone)]
struct CounterAccount {
    count: u64,
}

struct CounterDecoder;

impl<'a> AccountDecoder<'a> for CounterDecoder {
    type AccountType = CounterAccount;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<CounterAccount>> {
        if account.owner != COUNTER_PROGRAM_ID {
            return None;
        }
        let payload = account.data.strip_prefix(b"counter!")?;
        let count = u64::from_le_bytes(payload.get(..8)?.try_into().ok()?);

        Some(DecodedAccount {
            lamports: account.lamports,
            data: CounterAccount { count },
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct CounterLogger;

impl Processor<AccountProcessorInputType<'_, CounterAccount>> for CounterLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, CounterAccount>,
    ) -> CarbonResult<()> {
        println!(
            "slot {}: {} count={}",
            input.metadata.slot, input.metadata.pubkey, input.decoded_account.data.count
        );
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    let token = CancellationToken::new();
    let stop = token.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(3_500)).await;
        stop.cancel();
    });

    Pipeline::builder()
        .datasource(Ticker)
        .account(CounterDecoder, CounterLogger)
        .datasource_cancellation_token(token)
        .build()?
        .run()
        .await
}

Terminal

cargo run

Output

slot 1: 11157t3sqMV725NVRLrVQbAu98Jjfk1uCKehJnXXQs count=1
slot 2: 1117mWrzzrZr312ebPDHu8tbfMwFNvCvMbr6WepCNG count=2
slot 3: 1119DWteoLSdjvrT6g6L8C2PfDD2faiTQUpsjY2RiF count=3

Datasource#

Processor#

AccountDecoder#

InstructionDecoder#

CarbonDeserialize#

ArrangeAccounts#

InstructionDecoderCollection#

Filter#

MetricsExporter#

Metric#

Complete example#

`Datasource`#

`Processor`#

`AccountDecoder`#

`InstructionDecoder`#

`CarbonDeserialize`#

`ArrangeAccounts`#

`InstructionDecoderCollection`#

`Filter`#

`MetricsExporter`#

`Metric`#