stream

package
v1.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 23, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package stream provides Webull market-data streaming over MQTT.

A streaming client is built from the public client.Client, which supplies the credentials, region, and the resolved MQTT endpoints: 

cl, err := client.New(
	client.WithAppKey(key),
	client.WithAppSecret(secret),
	client.WithSandbox(),
)
if err != nil {
	return err
}
if _, err := cl.EnsureToken(ctx); err != nil {
	return err
}

s, err := stream.New(cl)
if err != nil {
	return err
}
defer s.Close()

s.OnQuote(func(q *marketdatav1.Quote) { ... })
if err := s.Connect(ctx); err != nil {
	return err
}
if err := s.Subscribe(ctx, stream.SubscribeRequest{
	Symbols:  []string{"AAPL"},
	Category: stream.CategoryUSStock,
	SubTypes: []stream.SubType{stream.SubTypeQuote, stream.SubTypeSnapshot},
}); err != nil {
	return err
}

Protocol

The MQTT connection carries only pushes: subscribe and unsubscribe are HTTP calls made with Client.Subscribe and Client.Unsubscribe. The MQTT CONNECT packet uses a freshly generated session id as the client id, the App Key as the user name, and an arbitrary password.

Incoming messages are routed by topic and decoded before the registered handler is invoked:

Reconnection

Webull does not restore subscriptions after a connection is lost, and the MQTT broker never restores them either. With WithAutoReconnect, the client detects the lost connection, reconnects, and then re-issues the HTTP subscribe calls for every active subscription before the Client.OnConnect handlers run. Re-subscription is idempotent: each active subscription is re-issued exactly once and a symbol subscribed twice is only restored once. Client.OnDisconnect fires when a live connection is lost; Client.OnConnect fires after the initial connection and after every successful reconnect; Client.Reconnecting reports an in-progress reconnect attempt.

Limits

Webull applies the following connection rules, which the SDK reflects where it can:

  • An App Key supports at most five concurrent MQTT connections. Exceeding the limit fails the connection with Webull error code 105. When that happens Client.Connect returns an error explaining the limit.
  • A new connection that reuses an existing session id disconnects the previous connection. New therefore generates a unique session id by default; only set WithSessionID when exclusivity is guaranteed.
  • After a disconnect the server retains the connection state for about one minute. When the limit is reached, wait roughly a minute before reconnecting rather than retrying immediately.
  • The server pushes at most three messages per second per connection. The client does not need to throttle; this only bounds the inbound rate.

Index

Constants

View Source 
const (
	// DefaultKeepAlive is the MQTT keep-alive interval.
	DefaultKeepAlive = imqtt.DefaultKeepAlive
	// DefaultConnectTimeout bounds a single MQTT connection attempt.
	DefaultConnectTimeout = imqtt.DefaultConnectTimeout
	// DefaultWriteTimeout bounds writing an MQTT control packet.
	DefaultWriteTimeout = imqtt.DefaultWriteTimeout
	// DefaultMessageChannelDepth is the inbound message buffer size.
	DefaultMessageChannelDepth = imqtt.DefaultMessageChannelDepth
	// DefaultMaxReconnectInterval caps the reconnect backoff.
	DefaultMaxReconnectInterval = imqtt.DefaultMaxReconnectInterval
	// DefaultResubscribeTimeout bounds the whole re-subscription sequence
	// issued after a reconnect.
	DefaultResubscribeTimeout = 30 * time.Second
)

Default streaming parameters. Most mirror the low-level MQTT defaults.

View Source 
const (
	// TopicQuote carries real-time order-book depth as protobuf.
	TopicQuote = "quote"
	// TopicSnapshot carries market snapshots as protobuf.
	TopicSnapshot = "snapshot"
	// TopicTick carries tick-by-tick trades as protobuf.
	TopicTick = "tick"
	// TopicNotice carries server notifications as JSON.
	TopicNotice = "notice"
	// TopicEcho is a heartbeat with a null payload and is ignored.
	TopicEcho = "echo"
)

MQTT topics published by the Webull streaming broker.

Variables

This section is empty.

Functions

This section is empty.

Types

type Category 

type Category string

Category is the Webull security type accepted by the streaming subscribe/unsubscribe API. 

const (
	// CategoryUSStock selects United States stocks.
	CategoryUSStock Category = "US_STOCK"
	// CategoryUSETF selects United States ETFs.
	CategoryUSETF Category = "US_ETF"
	// CategoryHKStock selects Hong Kong stocks.
	CategoryHKStock Category = "HK_STOCK"
	// CategoryCNStock selects China Mainland A-Shares (Stock Connect).
	CategoryCNStock Category = "CN_STOCK"
)

Supported security categories.

func (Category) Valid 

func (c Category) Valid() bool

Valid reports whether c is a category accepted by Webull.

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is a Webull market-data streaming client. It owns an MQTT connection and routes decoded pushes to the handlers registered with the On* methods.

A Client is safe for concurrent use. Handlers may be registered before or after Client.Connect. The underlying client.Client is owned by the caller and is not closed by Client.Close.

func New 

func New(cl *client.Client, opts ...Option) (*Client, error)

New returns a streaming client bound to cl. When no session id is supplied with WithSessionID or WithClientID, New generates a unique one. New does not open a connection; call Client.Connect for that.

The broker address is taken from cl's resolved endpoints unless overridden with WithMQTTURL; by default the plain TCP endpoint is used and WithWebSocket selects the WebSocket endpoint instead.

func (*Client) Close 

func (c *Client) Close() error

Close disconnects from the broker. It is idempotent and does not close the underlying client.Client.

func (*Client) Connect 

func (c *Client) Connect(ctx context.Context) error

Connect establishes the MQTT connection, blocking until it succeeds, fails, or ctx is done. The first successful connection fires the Client.OnConnect handlers.

Webull does not restore subscriptions after a connection is lost. When a connection is re-established (see WithAutoReconnect), the client automatically re-issues the HTTP subscribe calls for every active subscription before the Client.OnConnect handlers run, so callers do not need to subscribe again.

A failure caused by Webull code 105 (the App Key already has five concurrent connections) is reported with a message explaining the limit and the roughly one-minute server retention window.

func (*Client) IsConnected 

func (c *Client) IsConnected() bool

IsConnected reports whether the MQTT connection is currently live. It is false while a reconnect is in progress.

func (*Client) OnConnect 

func (c *Client) OnConnect(fn func())

OnConnect registers a handler invoked after a successful connection, including automatic reconnections.

func (*Client) OnDisconnect 

func (c *Client) OnDisconnect(fn func(error))

OnDisconnect registers a handler invoked when an established connection is lost, with the reason.

func (*Client) OnError 

func (c *Client) OnError(fn func(error))

OnError registers a handler for asynchronous errors, such as a lost connection, a failed decode, or an unknown topic.

func (*Client) OnNotice 

func (c *Client) OnNotice(fn func([]byte))

OnNotice registers a handler for server notifications. The payload is delivered as raw JSON because the notice topic is not protobuf encoded.

func (*Client) OnQuote 

func (c *Client) OnQuote(fn func(*marketdatav1.Quote))

OnQuote registers a handler for order-book pushes on the quote topic.

func (*Client) OnSnapshot 

func (c *Client) OnSnapshot(fn func(*marketdatav1.Snapshot))

OnSnapshot registers a handler for market-snapshot pushes.

func (*Client) OnTick 

func (c *Client) OnTick(fn func(*marketdatav1.Tick))

OnTick registers a handler for tick-by-tick pushes.

func (*Client) Reconnecting 

func (c *Client) Reconnecting() bool

Reconnecting reports whether the client is currently attempting to re-establish a connection that was lost. It is always false before the first connection and after Client.Close.

func (*Client) SessionID 

func (c *Client) SessionID() string

SessionID returns the session id used as the MQTT client id and in subscribe/unsubscribe calls. It is stable for the lifetime of the Client.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, req SubscribeRequest) error

Subscribe starts streaming data for the requested symbols with POST /market-data/streaming/subscribe. The MQTT connection must already be established with Client.Connect, and an access token must be available (see client.Client.EnsureToken).

A successful subscription is recorded in the client's registry. Webull does not restore subscriptions across reconnects, so when the MQTT connection is re-established the client re-issues the active subscriptions automatically (see WithAutoResubscribe); callers do not need to call Subscribe again after a reconnect. Subscribing the same symbol and data type twice is idempotent and does not create a duplicate registration.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ctx context.Context, req UnsubscribeRequest) error

Unsubscribe stops streaming data for the requested symbols with POST /market-data/streaming/unsubscribe. Set UnsubscribeRequest.UnsubscribeAll to cancel every subscription for the session.

A successful unsubscribe removes the matching entries from the client's registry, so an unsubscribed symbol is not re-issued after a reconnect.

type Option 

type Option func(*config)

Option configures a Client during New. Options are applied in order.

func WithAutoReconnect 

func WithAutoReconnect(auto bool) Option

WithAutoReconnect enables the client's background reconnect loop. When a lost connection is re-established, the client automatically re-issues the active subscriptions (see WithAutoResubscribe) because Webull does not restore them. Client.OnConnect handlers fire again after a successful reconnect.

func WithAutoResubscribe 

func WithAutoResubscribe(auto bool) Option

WithAutoResubscribe controls whether the client re-issues its active HTTP subscriptions after a reconnect. It is enabled by default and is only meaningful together with WithAutoReconnect. Re-subscription is idempotent and never duplicates a subscription.

func WithCleanSession 

func WithCleanSession(clean bool) Option

WithCleanSession requests a clean MQTT session. Streaming connections are stateless, so the default is true.

func WithClientID 

func WithClientID(id string) Option

WithClientID sets the MQTT client id independently of the session id. Webull requires the MQTT client id to equal the session id used in the subscribe/unsubscribe calls, so this is an alias for WithSessionID; it is provided for callers that express the connection in MQTT terms.

func WithConnectTimeout 

func WithConnectTimeout(d time.Duration) Option

WithConnectTimeout sets how long a single MQTT connection attempt may take. Non-positive values are ignored by New, which keeps the previous value.

func WithKeepAlive 

func WithKeepAlive(d time.Duration) Option

WithKeepAlive sets the MQTT keep-alive interval. Non-positive values are ignored by New, which keeps the previous value.

func WithMQTTURL 

func WithMQTTURL(rawURL string) Option

WithMQTTURL overrides the broker address resolved from the client endpoints. It accepts a paho URL (tcp://host:port, wss://host:port/path) or a bare host:port, which is treated as tcp://host:port.

func WithMessageChannelDepth 

func WithMessageChannelDepth(depth uint) Option

WithMessageChannelDepth sets the number of inbound messages buffered by the MQTT client before traffic is throttled. A zero value is ignored by New.

func WithResubscribeTimeout 

func WithResubscribeTimeout(d time.Duration) Option

WithResubscribeTimeout bounds the whole re-subscription sequence issued after a reconnect. Non-positive values are ignored by New, which keeps the default (DefaultResubscribeTimeout).

func WithSessionID 

func WithSessionID(id string) Option

WithSessionID sets the session id used both as the MQTT client id and in the subscribe/unsubscribe calls. When unset, New generates a unique id.

Do not reuse a session id across connections under one App Key: Webull disconnects the previous connection when a new one connects with the same session id. Reusing an id therefore silently drops another client's stream. The generated default is unique per New call; only override it when a stable id is required and connection exclusivity is guaranteed.

func WithTLSConfig 

func WithTLSConfig(tc *tls.Config) Option

WithTLSConfig overrides the TLS configuration used for WebSocket (wss) or TLS connections. A nil value lets the MQTT client use a default configuration.

func WithWebSocket 

func WithWebSocket(useWebSocket bool) Option

WithWebSocket selects the MQTT-over-WebSocket endpoint instead of the plain TCP endpoint. It is ignored when WithMQTTURL is also set.

func WithWriteTimeout 

func WithWriteTimeout(d time.Duration) Option

WithWriteTimeout sets the timeout for writing an MQTT control packet. Non-positive values are ignored by New, which keeps the previous value.

type SubType 

type SubType string

SubType is a streaming data type accepted by the subscribe/unsubscribe API. 

const (
	// SubTypeQuote selects real-time order-book data.
	SubTypeQuote SubType = "QUOTE"
	// SubTypeSnapshot selects market snapshots.
	SubTypeSnapshot SubType = "SNAPSHOT"
	// SubTypeTick selects tick-by-tick trades.
	SubTypeTick SubType = "TICK"
)

Supported subscription data types.

func (SubType) Valid 

func (s SubType) Valid() bool

Valid reports whether s is a data type accepted by Webull.

type SubscribeRequest 

type SubscribeRequest struct {
	// SessionID overrides the client's session id for this call. It must match
	// the session id of the live MQTT connection; leave it empty to use
	// [Client.SessionID].
	SessionID string
	// Symbols are the security symbols to subscribe to, for example "AAPL".
	// At most 100 symbols are allowed per request.
	Symbols []string
	// Category is the security type.
	Category Category
	// SubTypes are the data types to receive.
	SubTypes []SubType
	// Grab requests that a snapshot be pushed immediately on subscription.
	Grab bool
	// Depth is the level-2 order-book depth. It is optional and defaults to
	// the server's value (10 levels); US stocks support at most 50 levels.
	Depth string
	// OvernightRequired includes the overnight session for US stocks.
	OvernightRequired bool
}

SubscribeRequest is the body of an HTTP streaming subscribe call.

type UnsubscribeRequest 

type UnsubscribeRequest struct {
	// SessionID overrides the client's session id for this call. Leave it
	// empty to use [Client.SessionID].
	SessionID string
	// Symbols are the security symbols to unsubscribe from. They are ignored
	// when UnsubscribeAll is true.
	Symbols []string
	// Category is the security type. It is ignored when UnsubscribeAll is
	// true.
	Category Category
	// SubTypes are the data types to stop receiving. They are ignored when
	// UnsubscribeAll is true.
	SubTypes []SubType
	// UnsubscribeAll cancels every subscription for the session; when set,
	// the other selection fields may be empty.
	UnsubscribeAll bool
}

UnsubscribeRequest is the body of an HTTP streaming unsubscribe call.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.