Struct ProtocolStreamBuilder 

pub struct ProtocolStreamBuilder { /* private fields */ }

Builds and configures the multi protocol stream described in the module-level docs.

See the module documentation for details on protocols, configuration options, and stream behavior.

Implementations

impl ProtocolStreamBuilder

pub fn new(tycho_url: &str, chain: Chain) -> Self

Creates a new builder for a multi-protocol stream.

See the module-level docs for full details on stream behavior and configuration.

pub fn exchange<T>( self, name: &str, filter: ComponentFilter, filter_fn: Option<fn(&ComponentWithState) -> bool>, ) -> Self

where T: ProtocolSim + TryFromWithBlock<ComponentWithState, BlockHeader, Error = InvalidSnapshotError> + Send + 'static,

Adds a specific exchange to the stream.

This configures the builder to include a new protocol synchronizer for name, filtering its components according to filter and optionally filter_fn.

The type parameter T specifies the decoder type for this exchange. All component states for this exchange will be decoded into instances of T.

Parameters

• name: The protocol or exchange name (e.g., "uniswap_v4", "vm:balancer_v2").
• filter: Defines the set of components to include in the stream.
• filter_fn: Optional custom filter function for client-side filtering of components not expressible in filter.

Notes

For certain protocols (e.g., "uniswap_v4", "vm:balancer_v2", "vm:curve"), omitting filter_fn may cause decoding errors or incorrect results. In these cases, a proper filter function is required to ensure correct decoding and quoting logic.

pub fn exchange_with_decoder_context<T>( self, name: &str, filter: ComponentFilter, filter_fn: Option<fn(&ComponentWithState) -> bool>, decoder_context: DecoderContext, ) -> Self

where T: ProtocolSim + TryFromWithBlock<ComponentWithState, BlockHeader, Error = InvalidSnapshotError> + Send + 'static,

Adds a specific exchange to the stream with decoder context.

This configures the builder to include a new protocol synchronizer for name, filtering its components according to filter and optionally filter_fn. It also registers the DecoderContext (this is useful to test protocols that are not live yet)

The type parameter T specifies the decoder type for this exchange. All component states for this exchange will be decoded into instances of T.

Parameters

• name: The protocol or exchange name (e.g., "uniswap_v4", "vm:balancer_v2").
• filter: Defines the set of components to include in the stream.
• filter_fn: Optional custom filter function for client-side filtering of components not expressible in filter.
• decoder_context: The decoder context for this exchange

Notes

For certain protocols (e.g., "uniswap_v4", "vm:balancer_v2", "vm:curve"), omitting filter_fn may cause decoding errors or incorrect results. In these cases, a proper filter function is required to ensure correct decoding and quoting logic.

pub fn block_time(self, block_time: u64) -> Self

Sets the block time interval for the stream.

This controls how often the stream produces updates.

pub fn timeout(self, timeout: u64) -> Self

👎Deprecated:

Use latency_buffer instead

Sets the network operation timeout (deprecated).

Use latency_buffer() instead for controlling latency. This method is retained for backwards compatibility.

pub fn latency_buffer(self, timeout: u64) -> Self

Sets the latency buffer to aggregate same-block messages.

This allows the supervisor to wait a short interval for all synchronizers to emit before aggregating.

pub fn max_missed_blocks(self, n: u64) -> Self

Sets the maximum number of blocks a synchronizer may miss before being marked as Stale.

pub fn startup_timeout(self, timeout: Duration) -> Self

Sets how long a synchronizer may take to process the initial message.

Useful for data-intensive protocols where startup decoding takes longer.

pub fn no_state(self, no_state: bool) -> Self

Configures the stream to exclude state updates.

This reduces bandwidth and decoding workload if protocol state is not of interest (e.g. only process new tokens).

pub fn auth_key(self, auth_key: Option<String>) -> Self

Sets the API key for authenticating with the Tycho server.

pub fn no_tls(self, no_tls: bool) -> Self

Disables TLS/ SSL for the connection, using http and ws protocols.

This is not recommended for production use.

pub fn disable_compression(self) -> Self

Disable compression for the connection.

pub fn enable_partial_blocks(self) -> Self

Enables partial block updates (flashblocks).

pub fn blocklist_components(self, ids: HashSet<String>) -> Self

Exclude specific component IDs from all registered exchanges.

pub fn stream_end_policy(self, stream_end_policy: StreamEndPolicy) -> Self

Sets the stream end policy.

Controls when the stream should stop based on synchronizer states.

Note

The stream always ends latest if all protocols are stale or ended independent of this configuration. This allows you to end the stream earlier than that.

See self::StreamEndPolicy for possible configuration options.

pub async fn set_tokens(self, tokens: HashMap<Bytes, Token>) -> Self

Provides token metadata used to decode startup snapshots and initialize protocol states.

This is not a stream filter — components arriving after startup include their own token metadata. To restrict to specific tokens, filter in your consumer logic. New tokens arriving via stream deltas are added automatically if they meet the quality threshold.

pub fn skip_state_decode_failures(self, skip: bool) -> Self

Skips decoding errors for component state updates.

Allows the stream to continue processing even if some states fail to decode, logging a warning instead of panicking.

pub fn min_token_quality(self, quality: u32) -> Self

Sets the minimum token quality for tokens added via the stream.

Tokens arriving in stream deltas below this threshold are ignored. Defaults to 100. Set this to the same value used in load_all_tokens() to apply consistent filtering.

pub fn websocket_retry_config(self, config: &RetryConfiguration) -> Self

Configures the retry policy for websocket reconnects.

pub fn state_synchronizer_retry_config( self, config: &RetryConfiguration, ) -> Self

Configures the retry policy for state synchronization.

pub fn get_decoder(&self) -> &TychoStreamDecoder<BlockHeader>

pub fn with_pending_indexer( self, extractor: &str, indexer: Box<dyn TxDeltaIndexer>, ) -> Result<Self, StreamError>

Registers a TxDeltaIndexer for ephemeral pending-block simulation.

The indexer is associated with extractor (the protocol synchronizer name, e.g. "uniswap_v3"). Use build_with_pending to obtain both the confirmed stream and the pending processor.

Returns an error if extractor names a VM protocol (prefix "vm:"), which requires update_engine() and cannot be simulated natively.

pub fn with_step_controller(self) -> (Self, BlockStepController)

Enables controlled-step mode for testing.

Returns a BlockStepController that lets the caller decide when each buffered block is released for decoding. Call this before build or build_with_pending — both detect and wire up the gating automatically.

In production code, do not call this method; the stream runs at full speed.

pub async fn build_with_pending( self, ) -> Result<(impl Stream<Item = Result<Update, StreamDecodeError>>, PendingBlockProcessor), StreamError>

Builds the confirmed protocol stream and a PendingBlockProcessor that stays in sync with it automatically.

The stream pipeline forwards every confirmed FeedMessage to the processor via an internal unbounded channel — it never blocks waiting for the consumer. The consumer owns the returned PendingBlockProcessor exclusively and may wrap it in whatever synchronisation primitive suits their use case (e.g. Mutex for shared access, nothing for single-threaded use).

Call generate_pending_update to simulate a candidate bundle; it drains the channel automatically before computing.

pub async fn build( self, ) -> Result<impl Stream<Item = Result<Update, StreamDecodeError>>, StreamError>

Builds and returns the configured protocol stream.

See the module-level docs for details on stream behavior and emitted messages. This method applies all builder settings and starts the stream.