JetStream Client

Install

go get github.com/orbitflare/orbitflare-sdk-go

import "github.com/orbitflare/orbitflare-sdk-go/jetstream"

Building the client

client, err := jetstream.NewBuilder().
    URL("http://ny.jetstream.orbitflare.com").
    FallbackURL("http://fra.jetstream.orbitflare.com").
    Build()

The URL falls back to the ORBITFLARE_JETSTREAM_URL environment variable. All builder methods are identical to the gRPC client - same defaults, same behavior (Retry, Timeout, Keepalive, PingInterval, MaxMissedPongs, ChannelCapacity, Logger).

Building a subscription

JetStream supports transaction filters (no slots, blocks, or commitment - those are Yellowstone-specific). Each TransactionFilter exposes AccountInclude, AccountExclude, and AccountRequired.

request := jetstream.NewSubscribeRequestBuilder().
    Transactions("raydium", jetstream.NewTransactionFilter().
        AccountInclude("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8")).
    Transactions("pumpfun", jetstream.NewTransactionFilter().
        AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")).
    Build()

Reading the stream

Subscribe returns a *jetstream.Stream. Call Next(ctx):

stream := client.Subscribe(request)
defer stream.Close()

for {
    update, err := stream.Next(ctx)
    if errors.Is(err, io.EOF) {
        break
    }
    if err != nil {
        break
    }

    if tx := update.GetTransaction(); tx != nil {
        // tx.GetSlot() - the slot number
        // tx.GetTransaction() - transaction info with signature, account_keys, instructions
    }
}

Closing, multiple streams, reconnection, and ping/pong all work identically to the gRPC client.

Full example

A stream that watches Raydium AMM swaps and prints each transaction’s slot.

package main

import (
    "context"
    "errors"
    "fmt"
    "io"
    "log"

    "github.com/orbitflare/orbitflare-sdk-go/jetstream"
)

func main() {
    client, err := jetstream.NewBuilder().
        URL("http://ny.jetstream.orbitflare.com").
        Build()
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    request := jetstream.NewSubscribeRequestBuilder().
        Transactions("raydium", jetstream.NewTransactionFilter().
            AccountInclude("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8")).
        Build()

    stream := client.Subscribe(request)
    defer stream.Close()

    ctx := context.Background()
    var count int
    fmt.Println("streaming raydium txs...")

    for {
        update, err := stream.Next(ctx)
        if errors.Is(err, io.EOF) {
            return
        }
        if err != nil {
            log.Fatal(err)
        }
        if tx := update.GetTransaction(); tx != nil {
            count++
            fmt.Printf("#%d slot=%d\n", count, tx.GetSlot())
        }
    }
}

JetStream v2

JetStream v2 (github.com/orbitflare/orbitflare-sdk-go/jetstream/v2) runs on the same endpoints and authentication as v1 and is fully additive: v1 keeps working unchanged. What v2 adds:

Runtime-managed filters - add and remove filters on a live stream without reconnecting.
Per-message sequence numbers - every response carries a monotonic Sequence, so you can detect dropped messages.
Opt-in enrichment - request fee payer, program ids, compute-unit price, compute limit, resolved address-table addresses, and more, per transaction.
Slot lifecycle events - a separate server stream of slot alive/complete/dead events.

The v2 client lives under jetstream/v2. The builder is identical to v1 (same methods, defaults, ORBITFLARE_JETSTREAM_URL env var, and failover):

import jetstreamv2 "github.com/orbitflare/orbitflare-sdk-go/jetstream/v2"

client, err := jetstreamv2.NewBuilder().
    URL("http://ny.jetstream.orbitflare.com").
    FallbackURL("http://fra.jetstream.orbitflare.com").
    Build()

Building typed filters

Each filter is a TransactionFilter with a client-chosen id (via .WithID()). The id is echoed back on every matching transaction and in the filter-validation acknowledgement, so you can correlate matches and remove the filter later.

filter := jetstreamv2.NewTransactionFilter().
    AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P").
    IncludeEnrichment(true).
    WithID("pumpfun")

A filter must set at least one of AccountInclude, AccountExclude, or AccountRequired; empty (match-everything) filters are rejected. IncludeEnrichment is optional (default off) and applies to the whole subscription: if any active filter enables it, every transaction you receive is enriched.

Subscribing to transactions

stream := client.SubscribeTransactions(filter)
defer stream.Close()

for {
    resp, err := stream.Next(ctx)
    if errors.Is(err, io.EOF) {
        break
    }
    if err != nil {
        break
    }

    switch {
    case resp.GetTransaction() != nil:
        ft := resp.GetTransaction()
        // resp.GetSequence() - monotonic per-stream sequence number
        // ft.GetFilterIds() - which of your filters this tx matched
        if tx := ft.GetTransaction(); tx != nil {
            // tx.GetSlot(), tx.GetSignature(), tx.GetAccountKeys(), tx.GetInstructions()
            // enrichment (only when IncludeEnrichment is set): tx.GetFeePayer(),
            //   tx.GetProgramIds(), tx.GetComputeUnitPrice(), tx.GetComputeLimit(),
            //   tx.GetLoadedWritableAddresses(), tx.GetLoadedReadonlyAddresses()
            fmt.Printf("seq=%d slot=%d\n", resp.GetSequence(), tx.GetSlot())
        }
    case resp.GetFilterValidation() != nil:
        r := resp.GetFilterValidation() // one per filter after each add/remove
        fmt.Printf("filter %s accepted=%v %s\n", r.GetFilterId(), r.GetAccepted(), r.GetRejectionReason())
    case resp.GetHeartbeat() != nil:
        // resp.GetHeartbeat().GetServerTsMs() - server wall-clock, ms since epoch
    case resp.GetPong() != nil:
        // resp.GetPong().GetPingId()
    }
}

Managing filters on a live stream

Take a handle from the stream and add or remove filters without reconnecting. Removals reference the filter id you assigned with .WithID().

handle := stream.Handle()

handle.AddFilters([]*jetstreamv2.TxFilter{
    jetstreamv2.NewTransactionFilter().
        AccountInclude("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8").
        WithID("raydium"),
})

handle.RemoveFilters([]string{"pumpfun"})

stream.AddFilters(...) and stream.RemoveFilters(...) are shortcuts for the same thing. Each returns an error if the stream has been closed.

Slot lifecycle events

SubscribeSlots() is a separate server stream and takes no filters.

slots := client.SubscribeSlots()
defer slots.Close()

for {
    event, err := slots.Next(ctx)
    if errors.Is(err, io.EOF) {
        break
    }
    if err != nil {
        break
    }
    // event.GetSlot(), event.GetStatus() (a SlotStatus enum: alive / complete / dead),
    // event.GetParentSlot(), event.GetCurrentLeader(), event.GetSequence()
    fmt.Printf("slot=%d status=%s\n", event.GetSlot(), event.GetStatus())
}

Health probes

GetVersion(ctx) and Ping(ctx, count) are unary calls with the same failover as the streams.

version, err := client.GetVersion(ctx)
count, err := client.Ping(ctx, 7)

For the full protobuf specification, message shapes, enrichment fields, and slot statuses, see the JetStream v2 Protocol Reference.

Get Started

Agents

Products

Solutions

Fetching Data

RPC

Data Streaming

Trading APIs

Install

Building the client

Building a subscription

Reading the stream

Full example

JetStream v2

Building typed filters

Subscribing to transactions

Managing filters on a live stream

Slot lifecycle events

Health probes

​Install

​Building the client

​Building a subscription

​Reading the stream

​Full example

​JetStream v2

​Building typed filters

​Subscribing to transactions

​Managing filters on a live stream

​Slot lifecycle events

​Health probes

Install

Building the client

Building a subscription

Reading the stream

Full example

JetStream v2

Building typed filters

Subscribing to transactions

Managing filters on a live stream

Slot lifecycle events

Health probes