Go JetStream 客户端

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

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

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

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

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 的交易信息
    }
}

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

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

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

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

handle := stream.Handle()

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

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

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

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