Helius Laserstream

https://docs.rs/helius-laserstream/latest/
Admin
Helius Laserstream is a Rust SDK for subscribing to real-time Solana blockchain updates including...
Tokens:19,067
Snippets:189
Trust Score:9.1
Update:2 months ago
Show doc for...
Tokens:
Raw
### Using CompressedAccountFilterSet for Subscriptions

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/index.html

Demonstrates initializing a filter, adding pubkeys, and updating a subscription request when the filter state changes.

```rust
use helius_laserstream::cuckoo::{CompressedAccountFilterSet, Pubkey};
use helius_laserstream::grpc::SubscribeRequest;
let mut set = CompressedAccountFilterSet::with_capacity(2_000_000).unwrap();
for pk in tracked_pubkeys() { set.insert(pk).unwrap(); }

let mut req = SubscribeRequest::default();
set.insert_into_subscribe_request(&mut req, "tracked_accounts");
// ... subscribe(req) ...

// On change, mutate and re-send on the SAME stream:
set.insert(new_pk).unwrap();
if set.take_dirty() {
    set.insert_into_subscribe_request(&mut req, "tracked_accounts");
    // handle.write(req).await
}
```

--------------------------------

### Off-chain Client Program for PDA Derivation

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

Demonstrates how to derive a PDA off-chain and send a transaction to the on-chain program to initialize the vault account.

```rust
fn create_vault_account(
    client: &RpcClient,
    program_id: Address,
    payer: &Keypair,
) -> Result<()> {
    // Derive the PDA from the payer account, a string representing the unique
    // purpose of the account ("vault"), and the address of our on-chain program.
    let (vault_address, vault_bump_seed) = Address::find_program_address(
        &[b"vault", payer.pubkey().as_ref()],
        &program_id
    );

    // Get the amount of lamports needed to pay for the vault's rent
    let vault_account_size = usize::try_from(VAULT_ACCOUNT_SIZE)?;
    let lamports = client.get_minimum_balance_for_rent_exemption(vault_account_size)?;

    // The on-chain program's instruction data, imported from that program's crate.
    let instr_data = InstructionData {
        vault_bump_seed,
        lamports,
    };

    // The accounts required by both our on-chain program and the system program's
    // `create_account` instruction, including the vault's address.
    let accounts = vec![
        AccountMeta::new(payer.pubkey(), true),
        AccountMeta::new(vault_address, false),
        AccountMeta::new(solana_system_interface::program::ID, false),
    ];

    // Create the instruction by serializing our instruction data via borsh
    let instruction = Instruction::new_with_borsh(
        program_id,
        &instr_data,
        accounts,
    );

    let blockhash = client.get_latest_blockhash()?;

    let transaction = Transaction::new_signed_with_payer(
        &[instruction],
        Some(&payer.pubkey()),
        &[payer],
        blockhash,
    );

    client.send_and_confirm_transaction(&transaction)?;

    Ok(())
}
```

--------------------------------

### Subscription Lifecycle Management

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Establishes a gRPC connection and returns a stream of updates with automatic reconnection logic.

```rust
/// Establishes a gRPC connection, handles the subscription lifecycle,
/// and provides a stream of updates. Automatically reconnects on failure.
#[instrument(skip(config, request))]
pub fn subscribe(
    config: LaserstreamConfig,
    request: SubscribeRequest,
) -> (
    impl Stream<Item = Result<SubscribeUpdate, LaserstreamError>>,
    StreamHandle,
) {
    let (write_tx, mut write_rx) = mpsc::unbounded_channel::<SubscribeRequest>();
    let handle = StreamHandle { write_tx };
    let update_stream = stream! {
        let mut reconnect_attempts = 0;
        let mut tracked_slot: u64 = 0;

        // Determine the effective max reconnect attempts
        let effective_max_attempts = config
            .max_reconnect_attempts
            .unwrap_or(HARD_CAP_RECONNECT_ATTEMPTS) // Default to hard cap if not set
            .min(HARD_CAP_RECONNECT_ATTEMPTS); // Enforce hard cap

        // Keep original request for reconnection attempts
        let mut current_request = request.clone();
        let internal_slot_sub_id = format!("internal-{}", uuid::Uuid::new_v4().to_string().split('-').next().unwrap());
        
        // Get replay behavior from config
        let replay_enabled = config.replay;
        
        // Add internal slot subscription only when replay is enabled
        if replay_enabled {
            current_request.slots.insert(
                internal_slot_sub_id.clone(),
                SubscribeRequestFilterSlots {
                    filter_by_commitment: Some(true), // Use same commitment as user request
                    ..Default::default()
                }
            );
        }
        
        // Clear any user-provided from_slot if replay is disabled
        if !replay_enabled {
            current_request.from_slot = None;
        }
```

--------------------------------

### Implement Connection and Subscription Loop

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

The main loop handles reconnection, slot tracking for replay, and message processing. It uses tokio::select! to manage periodic pings and incoming stream updates.

```rust
        let api_key_string = config.api_key.clone();

        loop {
            // Drain any pending write requests that arrived during reconnection delay.
            // This ensures writes sent while disconnected are included in the next connection.
            while let Ok(write_request) = write_rx.try_recv() {
                merge_subscribe_requests(&mut current_request, &write_request, &internal_slot_sub_id);
            }

            // Always update from_slot on current_request based on tracked_slot.
            // This ensures reconnections always use the most recent slot, even after
            // a successful connection that subsequently errors on the stream.
            if tracked_slot > 0 && replay_enabled {
                let commitment_level = current_request.commitment.unwrap_or(0);
                let from_slot = match commitment_level {
                    0 => tracked_slot.saturating_sub(31), // PROCESSED: rewind by 31 slots
                    1 | 2 => tracked_slot,                 // CONFIRMED/FINALIZED: exact slot
                    _ => tracked_slot.saturating_sub(31),  // Unknown: default to safe behavior
                };
                current_request.from_slot = Some(from_slot);
            } else if !replay_enabled {
                current_request.from_slot = None;
            }

            let attempt_request = current_request.clone();

            match connect_and_subscribe_once(&config, attempt_request, api_key_string.clone()).await {
                Ok((sender, stream)) => {
                    // Successful connection – reset attempt counter so we don't hit the cap
                    reconnect_attempts = 0;

                    // Box sender and stream here before processing
                    let mut sender: Pin<Box<dyn futures_util::Sink<SubscribeRequest, Error = futures_mpsc::SendError> + Send>> = Box::pin(sender);
                    // Ensure the boxed stream yields Result<_, Status>
                    let mut stream: Pin<Box<dyn Stream<Item = Result<SubscribeUpdate, Status>> + Send>> = Box::pin(stream);

                    // Ping interval timer
                    let mut ping_interval = tokio::time::interval(Duration::from_secs(30));
                    ping_interval.tick().await; // Skip first immediate tick
                    let mut ping_id = 0i32;

                    loop {
                        tokio::select! {
                            // Send periodic ping
                            _ = ping_interval.tick() => {
                                ping_id = ping_id.wrapping_add(1);
                                let ping_request = SubscribeRequest {
                                    ping: Some(SubscribeRequestPing { id: ping_id }),
                                    ..Default::default()
                                };
                                let _ = sender.send(ping_request).await;
                            },
                            // Handle incoming messages from the server
                            result = stream.next() => {
                                if let Some(result) = result {
                                    match result {
                                        Ok(update) => {
                                            
                                            // Handle ping/pong
                                            if matches!(&update.update_oneof, Some(UpdateOneof::Ping(_))) {
                                                let pong_req = SubscribeRequest { ping: Some(SubscribeRequestPing { id: 1 }), ..Default::default() };
                                                if let Err(e) = sender.send(pong_req).await {
                                                    warn!(error = %e, "Failed to send pong");
                                                    break;
                                                }
                                                continue;
                                            }
                                            
                                            // Do not forward server 'Pong' updates to consumers either
                                            if matches!(&update.update_oneof, Some(UpdateOneof::Pong(_))) {
                                                continue;
                                            }

                                // Track the latest slot from any slot update (including internal subscription)
                                if let Some(UpdateOneof::Slot(s)) = &update.update_oneof {
                                    if replay_enabled {
                                        tracked_slot = s.slot;
                                    }
```

--------------------------------

### Configure and Connect Geyser Client

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Sets up the gRPC endpoint with TLS, timeouts, and compression settings before initiating a subscription.

```rust
#[instrument(skip(config, request, api_key))]
async fn connect_and_subscribe_preprocessed_once(
    config: &LaserstreamConfig,
    request: SubscribePreprocessedRequest,
    api_key: String,
) -> Result<
    impl Stream<Item = Result<SubscribePreprocessedUpdate, laserstream_core_proto::tonic::Status>> + Send,
    Status,
> {
    let options = &config.channel_options;

    // Create our custom interceptor with SDK metadata
    let interceptor = SdkMetadataInterceptor::new(api_key)?;

    // Build endpoint with all options
    let mut endpoint = Endpoint::from_shared(config.endpoint.clone())
        .map_err(|e| Status::internal(format!("Failed to parse endpoint: {}", e)))?
        .connect_timeout(Duration::from_secs(options.connect_timeout_secs.unwrap_or(10)))
        .timeout(Duration::from_secs(options.timeout_secs.unwrap_or(30)))
        .tcp_nodelay(options.tcp_nodelay.unwrap_or(true))
        .tcp_keepalive(Some(Duration::from_secs(options.tcp_keepalive_secs.unwrap_or(30))))
        .http2_keep_alive_interval(Duration::from_secs(options.http2_keep_alive_interval_secs.unwrap_or(30)))
        .keep_alive_timeout(Duration::from_secs(options.keep_alive_timeout_secs.unwrap_or(10)))
        .keep_alive_while_idle(options.keep_alive_while_idle.unwrap_or(true));

    endpoint = endpoint
        .tls_config(ClientTlsConfig::new().with_enabled_roots())
        .map_err(|e| Status::internal(format!("Failed to configure TLS: {}", e)))?;

    let channel = endpoint
        .connect()
        .await
        .map_err(|e| Status::internal(format!("Failed to connect: {}", e)))?;

    let mut geyser_client = GeyserClient::with_interceptor(channel, interceptor)
        .max_decoding_message_size(options.max_decoding_message_size.unwrap_or(1_000_000_000))
        .max_encoding_message_size(options.max_encoding_message_size.unwrap_or(64 * 1024 * 1024));

    // Apply compression if specified
    if let Some(compression) = &options.send_compression {
        let encoding = match compression {
            ConfigCompressionEncoding::Gzip => CompressionEncoding::Gzip,
            ConfigCompressionEncoding::Zstd => CompressionEncoding::Zstd,
        };
        geyser_client = geyser_client.send_compressed(encoding).accept_compressed(encoding);
    }

    let (mut subscribe_tx, subscribe_rx) = futures_mpsc::unbounded();

    subscribe_tx
        .send(request)
        .await
        .map_err(|e| Status::internal(format!("Failed to send initial request: {}", e)))?;

    let response = geyser_client
        .subscribe_preprocessed(subscribe_rx)
        .await
        .map_err(|e| Status::internal(format!("Preprocessed subscription failed: {}", e)))?;

    Ok(response.into_inner())
}
```

--------------------------------

### On-chain Solana Program for PDA Account Creation

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

Defines the instruction data structure and the entrypoint for creating a vault account owned by the program.

```rust
// The custom instruction processed by our program. It includes the
// PDA's bump seed, which is derived by the client program. This
// definition is also imported into the off-chain client program.
// The computed address of the PDA will be passed to this program via
// the `accounts` vector of the `Instruction` type.
#[derive(BorshSerialize, BorshDeserialize, Debug)]
pub struct InstructionData {
    pub vault_bump_seed: u8,
    pub lamports: u64,
}

// The size in bytes of a vault account. The client program needs
// this information to calculate the quantity of lamports necessary
// to pay for the account's rent.
pub static VAULT_ACCOUNT_SIZE: u64 = 1024;

// The entrypoint of the on-chain program, as provided to the
// `entrypoint!` macro.
fn process_instruction(
    program_id: &Address,
    accounts: &[AccountInfo],
    instruction_data: &[u8],
) -> ProgramResult {
    let account_info_iter = &mut accounts.iter();
    let payer = next_account_info(account_info_iter)?;
    // The vault PDA, derived from the payer's address
    let vault = next_account_info(account_info_iter)?;

    let mut instruction_data = instruction_data;
    let instr = InstructionData::deserialize(&mut instruction_data)?;
    let vault_bump_seed = instr.vault_bump_seed;
    let lamports = instr.lamports;
    let vault_size = VAULT_ACCOUNT_SIZE;

    // Invoke the system program to create an account while virtually
    // signing with the vault PDA, which is owned by this caller program.
    invoke_signed(
        &create_account(
            &payer.key,
            &vault.key,
            lamports,
            vault_size,
            program_id,
        ),
        &[
            payer.clone(),
            vault.clone(),
        ],
        // A slice of seed slices, each seed slice being the set
        // of seeds used to generate one of the PDAs required by the
        // callee program, the final seed being a single-element slice
        // containing the `u8` bump seed.
        &[
            &[
                b"vault",
                payer.key.as_ref(),
                &[vault_bump_seed],
            ],
        ]
    )?;

    Ok(())
}
```

--------------------------------

### Initialize and Configure Laserstream

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/config.rs.html

Methods for creating a new configuration instance and applying builder-pattern modifications.

```rust
impl Default for LaserstreamConfig {
    fn default() -> Self {
        Self {
            api_key: String::new(),
            endpoint: String::new(),
            max_reconnect_attempts: None, // Default to None
            channel_options: ChannelOptions::default(),
            replay: true, // Default to true
        }
    }
}

impl LaserstreamConfig {
    pub fn new(endpoint: String, api_key: String) -> Self {
        Self {
            endpoint,
            api_key,
            max_reconnect_attempts: None, // Default to None
            channel_options: ChannelOptions::default(),
            replay: true, // Default to true
        }
    }

    /// Sets the maximum number of reconnection attempts.
    pub fn with_max_reconnect_attempts(mut self, attempts: u32) -> Self {
        self.max_reconnect_attempts = Some(attempts);
        self
    }

    /// Sets custom channel options.
    pub fn with_channel_options(mut self, options: ChannelOptions) -> Self {
        self.channel_options = options;
        self
    }

    /// Sets replay behavior on reconnects.
    /// When true (default), uses from_slot and internal slot tracking for replay.
    /// When false, starts from current slot on reconnects (no replay).
    pub fn with_replay(mut self, replay: bool) -> Self {
        self.replay = replay;
        self
    }
}
```

--------------------------------

### Implement Compressed Account Filtering

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/lib.rs.html

Demonstrates using a Cuckoo filter to track large sets of pubkeys with minimal memory overhead. Requires the 'cuckoo' feature flag.

```rust
use helius_laserstream::cuckoo::{CompressedAccountFilterSet, Pubkey};
use helius_laserstream::grpc::SubscribeRequest;
# fn tracked_pubkeys() -> Vec<Pubkey> { vec![] }
# let new_pk = Pubkey::default();
let mut set = CompressedAccountFilterSet::with_capacity(2_000_000).unwrap();
for pk in tracked_pubkeys() { set.insert(pk).unwrap(); }

let mut req = SubscribeRequest::default();
set.insert_into_subscribe_request(&mut req, "tracked_accounts");
// ... subscribe(req) ...

// On change, mutate and re-send on the SAME stream:
set.insert(new_pk).unwrap();
if set.take_dirty() {
    set.insert_into_subscribe_request(&mut req, "tracked_accounts");
    // handle.write(req).await
}
```

--------------------------------

### Configure Channel Compression

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/config.rs.html

Helper methods to set Zstd or Gzip compression for gRPC channels.

```rust
impl ChannelOptions {
    /// Enable zstd compression for both sending and receiving
    pub fn with_zstd_compression(mut self) -> Self {
        self.send_compression = Some(CompressionEncoding::Zstd);
        self.accept_compression = Some(vec![CompressionEncoding::Zstd, CompressionEncoding::Gzip]);
        self
    }
    
    /// Enable gzip compression for both sending and receiving
    pub fn with_gzip_compression(mut self) -> Self {
        self.send_compression = Some(CompressionEncoding::Gzip);
        self.accept_compression = Some(vec![CompressionEncoding::Gzip, CompressionEncoding::Zstd]);
        self
    }
}
```

--------------------------------

### get_version

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Retrieves the server version.

```APIDOC
## gRPC get_version

### Description
Queries the server for its current version information.

### Parameters
- **request** (Request<GetVersionRequest>) - The request for the server version.
```

--------------------------------

### Subscribe to Preprocessed Transactions

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Initializes a stream for preprocessed transactions with automatic reconnection capabilities.

```rust
/// Handle for managing a preprocessed subscription (no write support).
#[derive(Clone)]
pub struct PreprocessedStreamHandle;

/// Establishes a gRPC connection for preprocessed transactions and provides a stream of updates.
/// Automatically reconnects on failure. No slot tracking or replay - just simple reconnection.
#[instrument(skip(config, request))]
pub fn subscribe_preprocessed(
    config: LaserstreamConfig,
    request: SubscribePreprocessedRequest,
) -> (
    impl Stream<Item = Result<SubscribePreprocessedUpdate, LaserstreamError>>,
    PreprocessedStreamHandle,
) {
    let handle = PreprocessedStreamHandle;
    let update_stream = stream! {
        let mut reconnect_attempts = 0;

        // Determine the effective max reconnect attempts
        let effective_max_attempts = config
            .max_reconnect_attempts
```

--------------------------------

### Verify a Program Derived Address

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

Demonstrates how to verify that a set of seeds and a bump seed correctly generate an expected program derived address.

```rust
let (expected_pda, bump_seed) = Address::find_program_address(&[b"vault"], &program_id);
let actual_pda = Address::create_program_address(&[b"vault", &[bump_seed]], &program_id)?;
assert_eq!(expected_pda, actual_pda);
```

--------------------------------

### subscribe(config: LaserstreamConfig, request: SubscribeRequest)

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/client/fn.subscribe.html

Establishes a gRPC connection, handles the subscription lifecycle, and provides a stream of updates. Automatically reconnects on failure.

```APIDOC
## subscribe

### Description
Establishes a gRPC connection, handles the subscription lifecycle, and provides a stream of updates. Automatically reconnects on failure.

### Signature
`pub fn subscribe(config: LaserstreamConfig, request: SubscribeRequest) -> (impl Stream<Item = Result<SubscribeUpdate, LaserstreamError>>, StreamHandle)`

### Parameters
- **config** (LaserstreamConfig) - Required - Configuration settings for the Laserstream client.
- **request** (SubscribeRequest) - Required - The subscription request parameters.

### Returns
- **(impl Stream<Item = Result<SubscribeUpdate, LaserstreamError>>, StreamHandle)** - A tuple containing the stream of updates and a handle to control the stream.
```

--------------------------------

### Initialize CuckooFilter with Default Hasher

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.CuckooFilter.html

Creates a filter sized for a specific capacity using the default wire-stable hasher.

```rust
pub fn with_capacity( max_capacity: usize, ) -> Result<CuckooFilter<T>, CuckooBuildError>
```

--------------------------------

### Connect and Subscribe to Laserstream

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Establishes a gRPC connection with custom interceptors, TLS, and channel options, returning a bidirectional stream for subscription requests and updates.

```rust
#[instrument(skip(config, request, api_key))]
async fn connect_and_subscribe_once(
    config: &LaserstreamConfig,
    request: SubscribeRequest,
    api_key: String,
) -> Result<
    (
        impl futures_util::Sink<SubscribeRequest, Error = futures_mpsc::SendError> + Send,
        impl Stream<Item = Result<SubscribeUpdate, laserstream_core_proto::tonic::Status>> + Send,
    ),
    Status,
> {
    let options = &config.channel_options;

    // Create our custom interceptor with SDK metadata
    let interceptor = SdkMetadataInterceptor::new(api_key)?;

    // Build endpoint with all options
    let mut endpoint = Endpoint::from_shared(config.endpoint.clone())
        .map_err(|e| Status::internal(format!("Failed to parse endpoint: {}", e)))?
        .connect_timeout(Duration::from_secs(options.connect_timeout_secs.unwrap_or(10)))
        .timeout(Duration::from_secs(options.timeout_secs.unwrap_or(30)))
        .http2_keep_alive_interval(Duration::from_secs(options.http2_keep_alive_interval_secs.unwrap_or(30)))
        .keep_alive_timeout(Duration::from_secs(options.keep_alive_timeout_secs.unwrap_or(5)))
        .keep_alive_while_idle(options.keep_alive_while_idle.unwrap_or(true))
        .initial_stream_window_size(options.initial_stream_window_size.or(Some(1024 * 1024 * 4)))
        .initial_connection_window_size(options.initial_connection_window_size.or(Some(1024 * 1024 * 8)))
        .http2_adaptive_window(options.http2_adaptive_window.unwrap_or(true))
        .tcp_nodelay(options.tcp_nodelay.unwrap_or(true))
        .buffer_size(options.buffer_size.or(Some(1024 * 64)));

    if let Some(tcp_keepalive_secs) = options.tcp_keepalive_secs {
        endpoint = endpoint.tcp_keepalive(Some(Duration::from_secs(tcp_keepalive_secs)));
    }

    // Configure TLS
    endpoint = endpoint
        .tls_config(ClientTlsConfig::new().with_enabled_roots())
        .map_err(|e| Status::internal(format!("TLS config error: {}", e)))?;

    // Connect to create channel
    let channel = endpoint
        .connect()
        .await
        .map_err(|e| Status::unavailable(format!("Connection failed: {}", e)))?;

    // Create geyser client with our custom interceptor
    let mut geyser_client = GeyserClient::with_interceptor(channel, interceptor);

    // Configure message size limits
    geyser_client = geyser_client
        .max_decoding_message_size(options.max_decoding_message_size.unwrap_or(1_000_000_000))
        .max_encoding_message_size(options.max_encoding_message_size.unwrap_or(64 * 1024 * 1024));

    // Configure compression if specified
    if let Some(send_comp) = options.send_compression {
        let encoding = match send_comp {
            ConfigCompressionEncoding::Gzip => CompressionEncoding::Gzip,
            ConfigCompressionEncoding::Zstd => CompressionEncoding::Zstd,
        };
        geyser_client = geyser_client.send_compressed(encoding);
    }

    // Configure accepted compression encodings
    if let Some(ref accept_comps) = options.accept_compression {
        for comp in accept_comps {
            let encoding = match comp {
                ConfigCompressionEncoding::Gzip => CompressionEncoding::Gzip,
                ConfigCompressionEncoding::Zstd => CompressionEncoding::Zstd,
            };
            geyser_client = geyser_client.accept_compressed(encoding);
        }
    }

    // Create bidirectional stream
    let (mut subscribe_tx, subscribe_rx) = futures_mpsc::unbounded();
    subscribe_tx
        .send(request)
        .await
        .map_err(|e| Status::internal(format!("Failed to send initial request: {}", e)))?;

    let response = geyser_client
        .subscribe(subscribe_rx)
        .await
        .map_err(|e| Status::internal(format!("Subscription failed: {}", e)))?;

    Ok((subscribe_tx, response.into_inner()))
}
```

--------------------------------

### Initialize CuckooFilter with Custom Hasher

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.CuckooFilter.html

Creates a filter with a custom hasher, which is intended for in-process use only and is not wire-compatible.

```rust
pub fn with_capacity_and_hasher( max_capacity: usize, hasher_builder: S, ) -> Result<CuckooFilter<T, S>, CuckooBuildError>
```

--------------------------------

### subscribe

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/client/index.html

Establishes a gRPC connection, handles the subscription lifecycle, and provides a stream of updates. Automatically reconnects on failure.

```APIDOC
## Function: subscribe

### Description
Establishes a gRPC connection, handles the subscription lifecycle, and provides a stream of updates. Automatically reconnects on failure.

### Returns
- **StreamHandle**: A handle for managing the bidirectional streaming subscription.
```

--------------------------------

### Define Laserstream Configuration and Channel Options

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/config.rs.html

Structures for configuring the Laserstream client, including authentication, endpoint, and gRPC channel parameters.

```rust
1use serde::{Deserialize, Serialize};

/// Compression encoding options
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CompressionEncoding {
    /// Gzip compression
    Gzip,
    /// Zstd compression
    Zstd,
}

#[derive(Debug, Clone)]
pub struct LaserstreamConfig {
    /// API Key for authentication.
    pub api_key: String,
    /// The Laserstream endpoint URL.
    pub endpoint: String,
    /// Maximum number of reconnection attempts. Defaults to 10.
    /// A hard cap of 240 attempts (20 minutes / 5 seconds) is enforced internally.
    pub max_reconnect_attempts: Option<u32>,
    /// gRPC channel options
    pub channel_options: ChannelOptions,
    /// When true, enable replay on reconnects (uses from_slot and internal slot tracking).
    /// When false, no replay - start from current slot on reconnects.
    /// Default: true
    pub replay: bool,
}

#[derive(Debug, Clone, Default)]
pub struct ChannelOptions {
    /// Connect timeout in seconds. Default: 10
    pub connect_timeout_secs: Option<u64>,
    /// Request timeout in seconds. Default: 30
    pub timeout_secs: Option<u64>,
    /// Max message size for receiving in bytes. Default: 1GB
    pub max_decoding_message_size: Option<usize>,
    /// Max message size for sending in bytes. Default: 32MB
    pub max_encoding_message_size: Option<usize>,
    /// HTTP/2 keep-alive interval in seconds. Default: 30
    pub http2_keep_alive_interval_secs: Option<u64>,
    /// Keep-alive timeout in seconds. Default: 5
    pub keep_alive_timeout_secs: Option<u64>,
    /// Enable keep-alive while idle. Default: true
    pub keep_alive_while_idle: Option<bool>,
    /// Initial stream window size in bytes. Default: 4MB
    pub initial_stream_window_size: Option<u32>,
    /// Initial connection window size in bytes. Default: 8MB
    pub initial_connection_window_size: Option<u32>,
    /// Enable HTTP/2 adaptive window. Default: true
    pub http2_adaptive_window: Option<bool>,
    /// Enable TCP no-delay. Default: true
    pub tcp_nodelay: Option<bool>,
    /// TCP keep-alive interval in seconds. Default: 60
    pub tcp_keepalive_secs: Option<u64>,
    /// Buffer size in bytes. Default: 64KB
    pub buffer_size: Option<usize>,
    /// Compression encodings to accept from server. Default: ["gzip", "zstd"]
    pub accept_compression: Option<Vec<CompressionEncoding>>,
    /// Compression encoding to use when sending. Default: None
    pub send_compression: Option<CompressionEncoding>,
}
```

--------------------------------

### subscribe

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Initiates a streaming subscription for blockchain updates.

```APIDOC
## gRPC subscribe

### Description
Establishes a streaming connection to receive updates based on the provided request.

### Parameters
- **request** (Request<Streaming<SubscribeRequest>>) - Required - The streaming request containing subscription criteria.
```

--------------------------------

### subscribe_preprocessed

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Initiates a streaming subscription for preprocessed blockchain updates.

```APIDOC
## gRPC subscribe_preprocessed

### Description
Establishes a streaming connection to receive preprocessed updates.

### Parameters
- **request** (Request<Streaming<SubscribePreprocessedRequest>>) - Required - The streaming request containing preprocessed subscription criteria.
```

--------------------------------

### Module Exports and Re-exports

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/lib.rs.html

The main library structure exposing client, configuration, and error modules, along with re-exported gRPC and Solana types.

```rust
pub mod client;
pub mod config;
pub mod error;

pub use client::{subscribe, subscribe_preprocessed, StreamHandle, PreprocessedStreamHandle};
pub use config::{ChannelOptions, LaserstreamConfig, CompressionEncoding};
pub use error::LaserstreamError;

// Re-export commonly used types from laserstream-core-proto
pub use laserstream_core_proto::geyser as grpc;
pub use laserstream_core_proto::solana;
```

--------------------------------

### ChannelOptions Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/config/struct.ChannelOptions.html

The structure containing configuration fields for gRPC channel management.

```rust
pub struct ChannelOptions {Show 15 fields
    pub connect_timeout_secs: Option<u64>,
    pub timeout_secs: Option<u64>,
    pub max_decoding_message_size: Option<usize>,
    pub max_encoding_message_size: Option<usize>,
    pub http2_keep_alive_interval_secs: Option<u64>,
    pub keep_alive_timeout_secs: Option<u64>,
    pub keep_alive_while_idle: Option<bool>,
    pub initial_stream_window_size: Option<u32>,
    pub initial_connection_window_size: Option<u32>,
    pub http2_adaptive_window: Option<bool>,
    pub tcp_nodelay: Option<bool>,
    pub tcp_keepalive_secs: Option<u64>,
    pub buffer_size: Option<usize>,
    pub accept_compression: Option<Vec<CompressionEncoding>>,
    pub send_compression: Option<CompressionEncoding>,
}
```

--------------------------------

### Write Subscription Request

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/client/struct.StreamHandle.html

Sends a new subscription request to update the active stream.

```rust
pub async fn write( &self, request: SubscribeRequest, ) -> Result<(), LaserstreamError>
```

--------------------------------

### subscribe

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/index.html

Function to initiate a subscription to a data stream.

```APIDOC
## subscribe

### Description
Initiates a subscription to a data stream using the provided configuration.

### Signature
`pub fn subscribe(config: LaserstreamConfig) -> Result<StreamHandle, LaserstreamError>`
```

--------------------------------

### Handle User Write Requests

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Processes user write requests by merging them with current subscription state to ensure internal slot tracking persists across reconnections.

```rust
// Handle write requests from the user
Some(write_request) = write_rx.recv() => {
    // Merge the write_request into current_request so it persists across reconnections
    merge_subscribe_requests(&mut current_request, &write_request, &internal_slot_sub_id);

    // Send the merged current_request (which preserves the internal slot
    // tracker) instead of the raw write_request. Yellowstone gRPC replaces
    // all subscriptions on each write, so the raw request would drop the
    // internal slot tracker and cause tracked_slot to go stale.
    let mut send_req = current_request.clone();
    send_req.from_slot = None;
    send_req.ping = None;

    if let Err(e) = sender.send(send_req).await {
        warn!(error = %e, "Failed to send write request");
        break;
    }
}
```

--------------------------------

### subscribe_preprocessed

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/client/index.html

Establishes a gRPC connection for preprocessed transactions and provides a stream of updates. Automatically reconnects on failure.

```APIDOC
## Function: subscribe_preprocessed

### Description
Establishes a gRPC connection for preprocessed transactions and provides a stream of updates. Automatically reconnects on failure. This method does not support slot tracking or replay.

### Returns
- **PreprocessedStreamHandle**: A handle for managing the preprocessed subscription.
```

--------------------------------

### From Trait Implementation

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.Transaction.html

Returns the argument unchanged.

```rust
fn from(t: T) -> T
```

--------------------------------

### subscribe_replay_info

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Retrieves replay information for a subscription.

```APIDOC
## gRPC subscribe_replay_info

### Description
Fetches replay information based on the provided request.

### Parameters
- **request** (Request<SubscribeReplayInfoRequest>) - The request containing replay information parameters.
```

--------------------------------

### Attempt to Derive Valid PDA

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

Iterates through bump seeds to find a valid program derived address.

```rust
pub fn derive_program_address<const N: usize>( seeds: &[&[u8]; N], program_id: &Address, ) -> Option<(Address, u8)>
```

--------------------------------

### LaserstreamConfig Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/config/struct.LaserstreamConfig.html

The primary configuration structure for initializing a Laserstream client.

```rust
pub struct LaserstreamConfig {
    pub api_key: String,
    pub endpoint: String,
    pub max_reconnect_attempts: Option<u32>,
    pub channel_options: ChannelOptions,
    pub replay: bool,
}
```

--------------------------------

### CuckooFilter::with_capacity_and_hasher

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.CuckooFilter.html

Creates a new CuckooFilter with a custom hasher builder.

```APIDOC
## pub fn with_capacity_and_hasher<S>(max_capacity: usize, hasher_builder: S) -> Result<CuckooFilter<T, S>, CuckooBuildError>

### Description
Creates a new CuckooFilter with a custom hasher. Note that custom hashers are in-process only and not wire-compatible.

### Parameters
- **max_capacity** (usize) - Required - The maximum number of items the filter is intended to hold.
- **hasher_builder** (S) - Required - The custom hasher builder to use.

### Returns
- **Result<CuckooFilter<T, S>, CuckooBuildError>** - Returns the initialized filter or a `CuckooBuildError::CapacityOverflow` if allocation fails.
```

--------------------------------

### subscribe_preprocessed

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream

Subscribes to a preprocessed data stream using the provided configuration.

```APIDOC
## subscribe_preprocessed

### Description
Subscribes to a preprocessed data stream.

### Signature
`pub fn subscribe_preprocessed(config: LaserstreamConfig) -> Result<PreprocessedStreamHandle, LaserstreamError>`
```

--------------------------------

### SDK Metadata Interceptor

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

A custom gRPC interceptor that injects authentication tokens and SDK versioning headers into outgoing requests.

```rust
/// Custom interceptor that adds SDK metadata headers to all gRPC requests
#[derive(Clone)]
struct SdkMetadataInterceptor {
    x_token: Option<laserstream_core_proto::tonic::metadata::AsciiMetadataValue>,
}

impl SdkMetadataInterceptor {
    fn new(api_key: String) -> Result<Self, Status> {
        let x_token = if !api_key.is_empty() {
            Some(api_key.parse().map_err(|e| {
                Status::invalid_argument(format!("Invalid API key: {}", e))
            })?)
        } else {
            None
        };
        Ok(Self { x_token })
    }
}

impl Interceptor for SdkMetadataInterceptor {
    fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
        // Add x-token if present
        if let Some(ref x_token) = self.x_token {
            request.metadata_mut().insert("x-token", x_token.clone());
        }

        // Add SDK metadata headers
        request.metadata_mut().insert("x-sdk-name", MetadataValue::from_static(SDK_NAME));
        request.metadata_mut().insert("x-sdk-version", MetadataValue::from_static(SDK_VERSION));

        Ok(request)
    }
}
```

--------------------------------

### subscribe_preprocessed

Source: https://docs.rs/helius-laserstream/latest/index.html

Subscribes to a preprocessed Helius Laserstream.

```APIDOC
## subscribe_preprocessed

### Description
Subscribes to a preprocessed Helius Laserstream.

### Signature
`pub fn subscribe_preprocessed(config: LaserstreamConfig) -> Result<PreprocessedStreamHandle, LaserstreamError>`
```

--------------------------------

### Pubkey Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

The basic structure definition for the Pubkey type.

```rust
pub struct Pubkey(/* private fields */);
```

--------------------------------

### CloneToUninit Trait Implementation

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.Transaction.html

Performs copy-assignment from self to a destination pointer. Requires nightly-only experimental API.

```rust
unsafe fn clone_to_uninit(&self, dest: *mut u8)
```

--------------------------------

### subscribe

Source: https://docs.rs/helius-laserstream/latest/index.html

Subscribes to a Helius Laserstream.

```APIDOC
## subscribe

### Description
Subscribes to a Helius Laserstream.

### Signature
`pub fn subscribe(config: LaserstreamConfig) -> Result<StreamHandle, LaserstreamError>`
```

--------------------------------

### Stream Handle Management

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Provides a handle to send subscription updates to an active bidirectional stream.

```rust
/// Handle for managing a bidirectional streaming subscription.
#[derive(Clone)]
pub struct StreamHandle {
    write_tx: mpsc::UnboundedSender<SubscribeRequest>,
}

impl StreamHandle {
    /// Send a new subscription request to update the active subscription.
    pub async fn write(&self, request: SubscribeRequest) -> Result<(), LaserstreamError> {
        self.write_tx
            .send(request)
            .map_err(|_| LaserstreamError::ConnectionError("Write channel closed".to_string()))
    }
}
```

--------------------------------

### CompiledInstruction Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.CompiledInstruction.html

The primary structure definition for CompiledInstruction, containing fields for program ID index, accounts, and data.

```rust
pub struct CompiledInstruction {
    pub program_id_index: u32,
    pub accounts: Vec<u8>,
    pub data: Vec<u8>,
}
```

--------------------------------

### UiTokenAmount Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.UiTokenAmount.html

The structure definition for UiTokenAmount, containing fields for the token amount and its decimal configuration.

```rust
pub struct UiTokenAmount {
    pub ui_amount: f64,
    pub decimals: u32,
    pub amount: String,
    pub ui_amount_string: String,
}
```

--------------------------------

### ReturnData Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.ReturnData.html

The structure definition for ReturnData, containing the program ID and data fields.

```rust
pub struct ReturnData {
    pub program_id: Vec<u8>,
    pub data: Vec<u8>,
}
```

--------------------------------

### Implement Reconnection Loop for Laserstream

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Handles stream reconnection attempts with a fixed delay and error logging.

```rust
        loop {
            let api_key = config.api_key.clone();
            let request_clone = request.clone();

            match connect_and_subscribe_preprocessed_once(&config, request_clone, api_key).await {
                Ok(mut stream) => {
                    reconnect_attempts = 0;

                    while let Some(result) = stream.next().await {
                        match result {
                            Ok(update) => yield Ok(update),
                            Err(e) => {
                                warn!(error = %e, "Stream error received");
                                break;
                            }
                        }
                    }
                }
                Err(err) => {
                    reconnect_attempts += 1;
                    error!(error = %err, attempt = reconnect_attempts, max_attempts = effective_max_attempts, "Connection failed, will retry after 5s delay");

                    if reconnect_attempts >= effective_max_attempts {
                        error!(attempts = effective_max_attempts, "Max reconnection attempts reached");
                        yield Err(LaserstreamError::MaxReconnectAttempts(Status::cancelled(
                            format!("Connection failed after {} attempts", effective_max_attempts)
                        )));
                        return;
                    }
                }
            }

            let delay = Duration::from_millis(FIXED_RECONNECT_INTERVAL_MS);
            sleep(delay).await;
        }
```

--------------------------------

### create_program_address

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

Creates a valid program derived address without searching for a bump seed, useful for verification.

```APIDOC
## pub fn create_program_address(seeds: &[&[u8]], program_id: &Address) -> Result<Address, AddressError>

### Description
Creates a valid program derived address without searching for a bump seed. It is primarily used for efficiently verifying that a set of seeds plus a bump seed derives a particular address.
```

--------------------------------

### get_slot

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Retrieves the current slot information.

```APIDOC
## gRPC get_slot

### Description
Queries the server for the current slot.

### Parameters
- **request** (Request<GetSlotRequest>) - The request for the current slot.
```

--------------------------------

### subscribe_preprocessed

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Initiates a server-side stream of preprocessed updates.

```APIDOC
## gRPC subscribe_preprocessed

### Description
Establishes a streaming connection to receive preprocessed updates from the Geyser server.

### Parameters
- **request** (Request<Streaming<SubscribePreprocessedRequest>>) - The stream of preprocessed subscription requests.
```

--------------------------------

### CompressedAccountFilterSet

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/lib.rs.html

Utility for tracking large sets of public keys using a compact cuckoo filter.

```APIDOC
## CompressedAccountFilterSet

### Description
Allows tracking large pubkey sets without re-uploading an explicit list every request. Uses a compact cuckoo filter (~3 bytes/account) to match against server-side data.

### Methods
- `with_capacity(capacity: usize) -> Result<Self, CuckooBuildError>`: Creates a new filter set with the specified capacity.
- `insert(pubkey: Pubkey) -> Result<(), TableFullError>`: Inserts a public key into the filter.
- `insert_into_subscribe_request(req: &mut SubscribeRequest, filter_name: &str)`: Adds the filter state to a subscription request.
- `take_dirty() -> bool`: Checks if the filter has been modified and resets the dirty flag.
```

--------------------------------

### CuckooFilter::with_capacity

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.CuckooFilter.html

Creates a new CuckooFilter sized for a specific number of items using the default hasher.

```APIDOC
## pub fn with_capacity(max_capacity: usize) -> Result<CuckooFilter<T>, CuckooBuildError>

### Description
Creates a new CuckooFilter instance sized to hold up to `max_capacity` items using the default `YellowstoneHasherBuilder`.

### Parameters
- **max_capacity** (usize) - Required - The maximum number of items the filter is intended to hold.

### Returns
- **Result<CuckooFilter<T>, CuckooBuildError>** - Returns the initialized filter or a `CuckooBuildError::CapacityOverflow` if allocation fails.
```

--------------------------------

### ping

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Performs a connectivity check.

```APIDOC
## gRPC ping

### Description
Sends a ping request to verify server availability.

### Parameters
- **request** (Request<PingRequest>) - The ping request object.
```

--------------------------------

### SubscribeUpdateBlockMeta Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/struct.SubscribeUpdateBlockMeta.html

The structure definition for SubscribeUpdateBlockMeta, detailing the fields available for block metadata.

```rust
pub struct SubscribeUpdateBlockMeta {
    pub slot: u64,
    pub blockhash: String,
    pub rewards: Option<Rewards>,
    pub block_time: Option<UnixTimestamp>,
    pub block_height: Option<BlockHeight>,
    pub parent_slot: u64,
    pub parent_blockhash: String,
    pub executed_transaction_count: u64,
    pub entries_count: u64,
}
```

--------------------------------

### subscribe

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Initiates a server-side stream of updates based on the provided SubscribeRequest.

```APIDOC
## gRPC subscribe

### Description
Establishes a streaming connection to receive updates from the Geyser server.

### Parameters
- **request** (Request<Streaming<SubscribeRequest>>) - The stream of subscription requests.
```

--------------------------------

### SubscribeRequest Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/struct.SubscribeRequest.html

The structure definition for SubscribeRequest, detailing the available filters and configuration options for gRPC subscriptions.

```rust
pub struct SubscribeRequest {
    pub accounts: HashMap<String, SubscribeRequestFilterAccounts>,
    pub slots: HashMap<String, SubscribeRequestFilterSlots>,
    pub transactions: HashMap<String, SubscribeRequestFilterTransactions>,
    pub transactions_status: HashMap<String, SubscribeRequestFilterTransactions>,
    pub blocks: HashMap<String, SubscribeRequestFilterBlocks>,
    pub blocks_meta: HashMap<String, SubscribeRequestFilterBlocksMeta>,
    pub entry: HashMap<String, SubscribeRequestFilterEntry>,
    pub commitment: Option<i32>,
    pub accounts_data_slice: Vec<SubscribeRequestAccountsDataSlice>,
    pub ping: Option<SubscribeRequestPing>,
    pub from_slot: Option<u64>,
}
```

--------------------------------

### Define SubscribeRequestFilterAccountsFilterMemcmp struct

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/struct.SubscribeRequestFilterAccountsFilterMemcmp.html

The structure definition for memory comparison filtering in account subscription requests.

```rust
pub struct SubscribeRequestFilterAccountsFilterMemcmp {
    pub offset: u64,
    pub data: Option<Data>,
}
```

--------------------------------

### Derive Program Address in Const Context

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/cuckoo/struct.Pubkey.html

A compile-time constant version of address derivation for use when all parameters are known at compile time.

```rust
pub const fn derive_address_const<const N: usize>( seeds: &[&[u8]; N], bump: Option<u8>, program_id: &Address, ) -> Address
```

--------------------------------

### Subscribe function signature

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/client/fn.subscribe.html

The function signature for establishing a gRPC subscription with automatic reconnection.

```rust
pub fn subscribe(
    config: LaserstreamConfig,
    request: SubscribeRequest,
) -> (impl Stream<Item = Result<SubscribeUpdate, LaserstreamError>>, StreamHandle)
```

--------------------------------

### Define MessageHeader Struct

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.MessageHeader.html

The structure definition for MessageHeader, containing fields for signature and account counts.

```rust
pub struct MessageHeader {
    pub num_required_signatures: u32,
    pub num_readonly_signed_accounts: u32,
    pub num_readonly_unsigned_accounts: u32,
}
```

--------------------------------

### TokenBalance Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.TokenBalance.html

The primary structure definition for TokenBalance, containing fields for account indexing and token details.

```rust
pub struct TokenBalance {
    pub account_index: u32,
    pub mint: String,
    pub ui_token_amount: Option<UiTokenAmount>,
    pub owner: String,
    pub program_id: String,
}
```

--------------------------------

### Into Trait Implementation

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/solana/storage/confirmed_block/struct.Transaction.html

Performs a conversion into another type U, provided U implements From<T>.

```rust
fn into(self) -> U
```

--------------------------------

### ping

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/geyser_server/trait.Geyser.html

Checks the connectivity of the server.

```APIDOC
## gRPC ping

### Description
Performs a health check or connectivity test against the server.

### Parameters
- **request** (Request<PingRequest>) - Required - The ping request object.
```

--------------------------------

### GetBlockHeightResponse Struct Definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/struct.GetBlockHeightResponse.html

The primary structure definition for the GetBlockHeightResponse message.

```rust
pub struct GetBlockHeightResponse {
    pub block_height: u64,
}
```

--------------------------------

### SubscribeUpdatePong struct definition

Source: https://docs.rs/helius-laserstream/latest/helius_laserstream/grpc/struct.SubscribeUpdatePong.html

The structure definition for the SubscribeUpdatePong message.

```rust
pub struct SubscribeUpdatePong {
    pub id: i32,
}
```

--------------------------------

### Process Stream Updates and Reconnection Logic

Source: https://docs.rs/helius-laserstream/latest/src/helius_laserstream/client.rs.html

Handles incoming stream updates, filters out internal subscription IDs when replay is enabled, and manages reconnection attempts upon stream errors.

```rust
// Skip if this slot update is EXCLUSIVELY from our internal subscription
if update.filters.len() == 1 && update.filters.contains(&internal_slot_sub_id) {
    continue;
}
}

// Filter out internal subscription from filters before yielding (only if replay is enabled)
let mut clean_update = update;
if replay_enabled {
    clean_update.filters.retain(|f| f != &internal_slot_sub_id);
    
    // Only yield if there are still filters after cleaning
    if !clean_update.filters.is_empty() {
        yield Ok(clean_update);
    }
} else {
    // When replay is disabled, yield all updates as-is
    yield Ok(clean_update);
}
}
Err(status) => {
// Transient error: reconnect silently. Surfaced to the consumer
// only on terminal failure (max attempts) — see the Err arm below.
warn!(error = %status, "Stream error, will reconnect after 5s delay");
break;
}
}
} else {
// Stream ended
break;
}
}
```
Tokens:
Raw