> ## 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 客户端 > OrbitFlare 解码后的碎片以带交易过滤器的 gRPC 流形式投递。 ## 安装 ```bash theme={null} go get github.com/orbitflare/orbitflare-sdk-go ``` ```go theme={null} import "github.com/orbitflare/orbitflare-sdk-go/jetstream" ``` ## 构建客户端 ```go theme={null} client, err := jetstream.NewBuilder(). URL("http://ny.jetstream.orbitflare.com"). FallbackURL("http://fra.jetstream.orbitflare.com"). Build() ``` URL 回退读取 `ORBITFLARE_JETSTREAM_URL` 环境变量。所有构建器方法与 [gRPC 客户端](/cn/sdk/go-grpc) 完全相同，默认值与行为一致（`Retry`、`Timeout`、`Keepalive`、`PingInterval`、`MaxMissedPongs`、`ChannelCapacity`、`Logger`）。 ## 构建订阅 JetStream 支持交易过滤器（不支持槽位、区块或承诺级别，那些是 Yellowstone 专有的）。每个 `TransactionFilter` 都提供 `AccountInclude`、`AccountExclude` 与 `AccountRequired`。 ```go theme={null} request := jetstream.NewSubscribeRequestBuilder(). Transactions("raydium", jetstream.NewTransactionFilter(). AccountInclude("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8")). Transactions("pumpfun", jetstream.NewTransactionFilter(). AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")). Build() ``` ## 读取流 `Subscribe` 返回一个 `*jetstream.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 } if tx := update.GetTransaction(); tx != nil { // tx.GetSlot() - 槽位号 // tx.GetTransaction() - 带签名、account_keys、instructions 的交易信息 } } ``` 关闭、多个流、重连与 ping/pong 的工作方式都与 [gRPC 客户端](/cn/sdk/go-grpc) 完全相同。 ## 完整示例 一个监听 Raydium AMM 兑换并打印每笔交易槽位的流。 ```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`）与 v1 运行在**相同的端点和身份验证**上，且完全向后兼容：v1 保持不变。v2 新增： * **运行时管理的过滤器** - 无需重连即可在活动流上添加与移除过滤器。 * **逐条消息的序列号** - 每条响应都带有单调递增的 `Sequence`，便于检测丢失的消息。 * **可选富化** - 可按交易请求手续费支付方、程序 id、计算单元价格、计算上限、解析后的地址表地址等。 * **槽位生命周期事件** - 独立的服务器流，投递槽位 alive/complete/dead 事件。 v2 客户端位于 `jetstream/v2` 下。构建器与 v1 完全相同（相同方法、默认值、`ORBITFLARE_JETSTREAM_URL` 环境变量与故障转移）： ```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() ``` ### 构建类型化过滤器 每个过滤器都是带有客户端自选 id（通过 `.WithID()`）的 `TransactionFilter`。该 id 会在每笔匹配的交易及过滤器校验确认中回传，便于你关联匹配结果并在之后移除该过滤器。 ```go theme={null} filter := jetstreamv2.NewTransactionFilter(). AccountInclude("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"). IncludeEnrichment(true). WithID("pumpfun") ``` 过滤器必须至少设置 `AccountInclude`、`AccountExclude` 或 `AccountRequired` 之一；空（匹配一切）的过滤器会被拒绝。`IncludeEnrichment` 为可选（默认关闭）且作用于整个订阅：只要任一活动过滤器启用了它，你收到的每笔交易都会被富化。 ### 订阅交易 ```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() - 单调递增的逐流序列号 // ft.GetFilterIds() - 本交易匹配了你的哪些过滤器 if tx := ft.GetTransaction(); tx != nil { // tx.GetSlot(), tx.GetSignature(), tx.GetAccountKeys(), tx.GetInstructions() // 富化（仅在设置了 IncludeEnrichment 时）：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() // 每次添加/移除后每个过滤器一条 fmt.Printf("filter %s accepted=%v %s\n", r.GetFilterId(), r.GetAccepted(), r.GetRejectionReason()) case resp.GetHeartbeat() != nil: // resp.GetHeartbeat().GetServerTsMs() - 服务器挂钟时间，自纪元起的毫秒数 case resp.GetPong() != nil: // resp.GetPong().GetPingId() } } ``` ### 在活动流上管理过滤器 从流中取得一个句柄，无需重连即可添加或移除过滤器。移除时引用你用 `.WithID()` 指定的过滤器 id。 ```go theme={null} handle := stream.Handle() handle.AddFilters([]*jetstreamv2.TxFilter{ jetstreamv2.NewTransactionFilter(). AccountInclude("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"). WithID("raydium"), }) handle.RemoveFilters([]string{"pumpfun"}) ``` `stream.AddFilters(...)` 与 `stream.RemoveFilters(...)` 是同一操作的快捷方式。若流已关闭，两者都会返回错误。 ### 槽位生命周期事件 `SubscribeSlots()` 是独立的服务器流，不接受过滤器。 ```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()（SlotStatus 枚举：alive / complete / dead）, // event.GetParentSlot(), event.GetCurrentLeader(), event.GetSequence() fmt.Printf("slot=%d status=%s\n", event.GetSlot(), event.GetStatus()) } ``` ### 健康探测 `GetVersion(ctx)` 与 `Ping(ctx, count)` 是一元调用，具备与流相同的故障转移。 ```go theme={null} version, err := client.GetVersion(ctx) count, err := client.Ping(ctx, 7) ``` 关于完整的 protobuf 规范、消息结构、富化字段及槽位状态，参见 [JetStream v2 协议参考](/cn/data-streaming/jetstream-v2-reference)。