> ## 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-клиенту](/ru/sdk/go-grpc) - те же значения по умолчанию и поведение (`Retry`, `Timeout`, `Keepalive`, `PingInterval`, `MaxMissedPongs`, `ChannelCapacity`, `Logger`). ## Построение подписки JetStream поддерживает фильтры транзакций (без слотов, блоков и commitment - это специфично для 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() - информация о транзакции с signature, account_keys, instructions } } ``` Закрытие, несколько потоков, переподключение и ping/pong работают точно так же, как у [gRPC-клиента](/ru/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 программ, цены compute unit, лимита вычислений, разрешённых адресов из таблиц адресов и др. на каждую транзакцию. * **События жизненного цикла слота** - отдельный серверный поток событий слота 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() ``` ### Построение типизированных фильтров Каждый фильтр - это `TransactionFilter` с выбранным клиентом id (через `.WithID()`). Этот 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() } } ``` ### Управление фильтрами на активном потоке Возьмите у потока хендл и добавляйте или удаляйте фильтры без переподключения. Удаление ссылается на id фильтра, заданный через `.WithID()`. ```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](/ru/data-streaming/jetstream-v2-reference).