gRPC 客户端（Yellowstone）

安装

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

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

构建客户端

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

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()

最小化设置：

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://。

.URL("http://ny.rpc.orbitflare.com:10000")

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

.URLs("http://ny.rpc.orbitflare.com:10000", "http://fra.rpc.orbitflare.com:10000")

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

.FallbackURL("http://fra.rpc.orbitflare.com:10000")

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

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

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

.Timeout(15 * time.Second)

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

.Keepalive(30 * time.Second)

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

.PingInterval(15 * time.Second)

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

.MaxMissedPongs(5)

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

.ChannelCapacity(8192)

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

构建订阅

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

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) 获取下一条更新：

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 消息在内部消费，绝不会出现在你的流中。

关闭

stream.Close()

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

多个流

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

pumpfun := client.Subscribe(pumpfunReq)
raydium := client.Subscribe(raydiumReq)
slots := client.Subscribe(slotsReq)

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

完整示例

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

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())
        }
    }
}

开始使用

产品

解决方案

数据获取

RPC

数据流

交易 API

gRPC 客户端（Yellowstone）

安装

构建客户端

构建器方法

构建订阅

过滤器类型

读取流

关闭

多个流

完整示例

​安装

​构建客户端

​构建器方法

​构建订阅

​过滤器类型

​读取流

​关闭

​多个流

​完整示例

安装

构建客户端

构建器方法

构建订阅

过滤器类型

读取流

关闭

多个流

完整示例