> ## Documentation Index
> Fetch the complete documentation index at: https://docs.orbitflare.com/llms.txt
> Use this file to discover all available pages before exploring further.

# JetStream Client

> OrbitFlare's decoded shreds delivered as gRPC streams with transaction filters.

## Install

```bash theme={null}
go get github.com/orbitflare/orbitflare-sdk-go
```

```go theme={null}
import "github.com/orbitflare/orbitflare-sdk-go/jetstream"
```

## Building the client

```go theme={null}
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](/sdk/go-grpc) - 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`.

```go theme={null}
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)`:

```go theme={null}
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](/sdk/go-grpc).

## Full example

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

```go theme={null}
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):

```go theme={null}
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.

```go theme={null}
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

```go theme={null}
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()`.

```go theme={null}
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.

```go theme={null}
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.

```go theme={null}
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](/data-streaming/jetstream-v2-reference).