> ## 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.

# gRPC 客户端（Yellowstone）

> 带自动重连、ping/pong 存活检测及类型化过滤器的 Yellowstone Geyser 流式传输。

## 安装

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

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

## 构建客户端

下面是设置了所有选项的客户端：

```go theme={null}
import (
    "time"

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

client, err := geyser.NewBuilder().
    URL("http://ny.rpc.orbitflare.com:10000").
    FallbackURL("http://fra.rpc.orbitflare.com:10000").
    Retry(orbitflare.RetryPolicy{
        InitialDelay: 100 * time.Millisecond,
        MaxDelay:     30 * time.Second,
        Multiplier:   2.0,
        MaxAttempts:  0,
    }).
    Timeout(30 * time.Second).
    Keepalive(60 * time.Second).
    PingInterval(10 * time.Second).
    MaxMissedPongs(3).
    ChannelCapacity(4096).
    Build()
```

最小化设置：

```go theme={null}
client, err := geyser.NewBuilder().
    URL("http://ny.rpc.orbitflare.com:10000").
    Build()
```

gRPC 无需 API 密钥。URL 回退读取 `ORBITFLARE_GRPC_URL` 环境变量。

### 构建器方法

**`.URL(url)`** - 主 gRPC 端点。回退读取 `ORBITFLARE_GRPC_URL`。明文 HTTP/2 使用 `http://`，TLS 使用 `https://`。

```go theme={null}
.URL("http://ny.rpc.orbitflare.com:10000")
```

**`.URLs(urls...)`** - 一次设置主端点与回退端点。第一个为主端点。

```go theme={null}
.URLs("http://ny.rpc.orbitflare.com:10000", "http://fra.rpc.orbitflare.com:10000")
```

**`.FallbackURL(url)` / `.FallbackURLs(urls...)`** - 添加回退端点。连接失败时，SDK 会轮换尝试。

```go theme={null}
.FallbackURL("http://fra.rpc.orbitflare.com:10000")
```

**`.Retry(policy)`** - 控制重连退避。连接断开时，SDK 等待 `InitialDelay`，随后每次尝试按倍数增长（上限为 `MaxDelay`）。将 `MaxAttempts` 设为 0 表示无限重试。默认：100ms 初始、30s 上限、2 倍系数、无限次。

```go theme={null}
.Retry(orbitflare.RetryPolicy{
    InitialDelay: 200 * time.Millisecond,
    MaxDelay:     15 * time.Second,
    Multiplier:   2.0,
    MaxAttempts:  0,
})
```

**`.Timeout(duration)`** - 连接/拨号超时。默认 30s。

```go theme={null}
.Timeout(15 * time.Second)
```

**`.Keepalive(duration)`** - TCP/HTTP2 keepalive 间隔，用于在传输层检测死连接。默认 60s。

```go theme={null}
.Keepalive(30 * time.Second)
```

**`.PingInterval(duration)`** - SDK 向服务器发送协议级 `Ping` 消息的频率。服务器应回以 `Pong`。默认 10s。

```go theme={null}
.PingInterval(15 * time.Second)
```

**`.MaxMissedPongs(n)`** - 连续多少次 ping 未获响应后，SDK 判定连接已死并重连。默认 3。在默认值下，死连接会在 30 秒内被检测到。

```go theme={null}
.MaxMissedPongs(5)
```

**`.ChannelCapacity(n)`** - 后台 goroutine 与你的代码之间的有界通道容量。若你的代码消费事件较慢，此通道满时后台 goroutine 会阻塞，而非无限占用内存。默认 4096。

```go theme={null}
.ChannelCapacity(8192)
```

**`.Logger(logger)`** - 用于连接生命周期事件的 `*slog.Logger`。未设置时由 `ORBITFLARE_LOG` 控制。

## 构建订阅

过滤器通过 `SubscribeRequestBuilder` 与类型化过滤器类型以编程方式构建。交易只要涉及任一 `AccountInclude` 地址即匹配；`AccountExclude` 排除匹配；`AccountRequired` 要求全部所列地址都出现。

```go theme={null}
request := geyser.NewSubscribeRequestBuilder().
    Transactions("pumpfun", geyser.NewTransactionFilter().
        Vote(false).
        Failed(false).
        AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")).
    Accounts("usdc-holders", geyser.NewAccountFilter().
        Owner("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA").
        Datasize(165).
        Memcmp(geyser.MemcmpBase58(0, "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v")).
        Lamports(geyser.LamportsGt(0))).
    Slots("slots", geyser.NewSlotFilter().FilterByCommitment(true)).
    Commitment(geyser.Confirmed).
    Build()
```

### 过滤器类型

* **`NewTransactionFilter()`** - `Vote(bool)`、`Failed(bool)`、`Signature(string)`、`AccountInclude(...)`、`AccountExclude(...)`、`AccountRequired(...)`。
* **`NewAccountFilter()`** - `Account(...)`、`Owner(...)`、`Datasize(uint64)`、`TokenAccountState(bool)`、`Memcmp(...)`、`Lamports(...)`、`NonemptyTxnSignature(bool)`。
* **`NewSlotFilter()`** - `FilterByCommitment(bool)`、`InterslotUpdates(bool)`。
* **`NewBlockFilter()`** - `AccountInclude(...)`、`IncludeTransactions(bool)`、`IncludeAccounts(bool)`、`IncludeEntries(bool)`。
* **`NewDeshredTransactionFilter()`** - `Vote(bool)`、`AccountInclude(...)`、`AccountExclude(...)`、`AccountRequired(...)`。

`Memcmp` 通过 `geyser.MemcmpBytes(offset, []byte)`、`geyser.MemcmpBase58(offset, string)` 或 `geyser.MemcmpBase64(offset, string)` 构建。`Lamports` 通过 `geyser.LamportsEq/Ne/Lt/Gt(uint64)` 构建。`Commitment` 为 `geyser.Processed`、`geyser.Confirmed`、`geyser.Finalized` 之一。

若你更愿意手动构建，`Subscribe` 也接受原始的 `*geyser.SubscribeRequest`（proto 类型的再导出）。

## 读取流

`Subscribe` 返回一个 `*geyser.Stream`。调用 `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 // 致命错误：所有重试已耗尽
    }

    switch {
    case update.GetTransaction() != nil:
        tx := update.GetTransaction()
        // tx.GetSlot(), tx.GetTransaction()（签名、账户、指令、meta）
    case update.GetAccount() != nil:
        acct := update.GetAccount()
        // acct.GetSlot(), acct.GetAccount()（pubkey、lamports、owner、data）, acct.GetIsStartup()
    case update.GetSlot() != nil:
        slot := update.GetSlot()
        // slot.GetSlot(), slot.GetStatus()
    case update.GetBlockMeta() != nil:
        meta := update.GetBlockMeta()
        // meta.GetSlot(), meta.GetBlockhash(), meta.GetParentSlot()
    }
}
```

`Next(ctx)` 会阻塞直到有事件到达。流结束后（包括 `Close` 之后）返回 `io.EOF`；任何其他非 nil 错误对该流都是致命的。若 context 被取消，它还会返回 `ctx.Err()`。Pong 消息在内部消费，绝不会出现在你的流中。

### 关闭

```go theme={null}
stream.Close()
```

立即停止后台 goroutine。`client.Close()` 会停止该客户端启动的每一个流。

## 多个流

一个客户端可同时运行多个流。每个流拥有各自的后台连接：

```go theme={null}
pumpfun := client.Subscribe(pumpfunReq)
raydium := client.Subscribe(raydiumReq)
slots := client.Subscribe(slotsReq)
```

它们共享端点健康状态。若某个流将某个失败端点隔离，其他流在下次重连时会跳过它。但它们的连接与生命周期完全独立。你可以随时启动新的流。

## 完整示例

一个监听 pump.fun 交易并为每笔打印摘要的流。

```go theme={null}
package main

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

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

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

    request := geyser.NewSubscribeRequestBuilder().
        Transactions("pumpfun", geyser.NewTransactionFilter().
            Vote(false).
            Failed(false).
            AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")).
        Slots("slots", geyser.NewSlotFilter().FilterByCommitment(true)).
        Commitment(geyser.Confirmed).
        Build()

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

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

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

        switch {
        case update.GetTransaction() != nil:
            count++
            tx := update.GetTransaction()
            fmt.Printf("#%d slot=%d\n", count, tx.GetSlot())
        case update.GetSlot() != nil:
            s := update.GetSlot()
            fmt.Printf("slot %d (%s)\n", s.GetSlot(), s.GetStatus())
        }
    }
}
```