Help us shape the future of Go! Take the 2026 Go Developer Survey

centrifuge

package module
v0.13.0 Latest Latest
Warning

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

 Go to latest Published: Oct 21, 2026 License: MIT Imports: 22 Imported by: 64

README

GoDoc

Websocket client for Centrifugo server and Centrifuge library.

There is no v1 release of this library yet – API still evolves. At the moment patch version updates only contain backwards compatible changes, minor version updates can have backwards incompatible API changes.

Check out client SDK API specification to learn how this SDK behaves. It's recommended to read that before starting to work with this SDK as the spec covers common SDK behavior - describes client and subscription state transitions, main options and methods. Also check out examples folder.

The features implemented by this SDK can be found in SDK feature matrix.

The latest centrifuge-go is compatible with Centrifugo server v6, v5 and v4, and Centrifuge >= 0.25.0. For Centrifugo v2, Centrifugo v3 and Centrifuge < 0.25.0 you should use centrifuge-go v0.8.x.

Callbacks should not block

When using this SDK you should not block for a long time inside event handlers since handlers called synchronously by the SDK and block the connection read loop. The fact that the read loop is blocked also means that you can not issue blocking Client requests such as Publish, RPC, History, Presence, PresenceStats from the event handler code – this will result into a deadlock. Use a separate goroutine if you really need to issue a blocking call from inside an event handler.

I.e. this code is broken:

client.OnMessage(func(e centrifuge.MessageEvent) {
    result, err := c.RPC(context.Background(), "method", []byte("{}"))
    if err != nil {
        log.Println(err)
        return
    }
    // Will never be called.
    log.Printf("RPC result: %s", string(result.Data))
})

This code is correct as it does not block event handler forever:

client.OnMessage(func(e centrifuge.MessageEvent) {
    // When issue blocking requests from inside event handler we must use
    // a goroutine. Otherwise, we will ge a deadlock since the connection
    // read loop is blocked.
    go func() {
        result, err := c.RPC(context.Background(), "method", []byte("{}"))
        if err != nil {
            log.Println(err)
            return
        }
        log.Printf("RPC result: %s", string(result.Data))
    }()
})

You can find similar limitations in eclipse/paho.mqtt.golang. In short, this is caused by a challenging mix of asynchronous protocol, Go and callback approach. In the previous versions of this SDK we allowed blocking requests from within event handlers – but it contradicts with the real-time nature of Centrifugal ecosystem, because we had to use separate callback queue, and that queue could grow huge without a reasonable way to backpressure (which we were not able to find).

If you are calling Publish, RPC, History, Presence, PresenceStats from the outside of event handler – you should not do any special care. Also, if you are calling your own blocking APIs from inside Centrifuge event handlers – you won't get the deadlock, but the read loop of the underlying connection will not proceed till the event handler returns.

Run tests

Start Centrifugo and run tests:

docker compose up -d
go test -race -v
docker compose down

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrTimeout returned if operation timed out.
	ErrTimeout = errors.New("timeout")
	// ErrClientDisconnected can be returned if client goes to
	// disconnected state while operation in progress.
	ErrClientDisconnected = errors.New("client disconnected")
	// ErrClientClosed can be returned if client is closed.
	ErrClientClosed = errors.New("client closed")
	// ErrSubscriptionUnsubscribed returned if Subscription is unsubscribed.
	ErrSubscriptionUnsubscribed = errors.New("subscription unsubscribed")
	// ErrDuplicateSubscription returned if subscription to the same channel
	// already registered in current client instance. This is due to the fact
	// that server does not allow subscribing to the same channel twice for
	// the same connection.
	ErrDuplicateSubscription = errors.New("duplicate subscription")
	// ErrUnauthorized is a special error which may be returned by application
	// from GetToken function to indicate lack of operation permission.
	ErrUnauthorized = errors.New("unauthorized")
)

Functions

This section is empty.

Types

type Client 

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

Client represents client connection to Centrifugo or Centrifuge library based server. It provides methods to set various event handlers, subscribe channels, call RPC commands etc. Call client Connect method to trigger actual connection with a server. Call client Close method to clean up state when you don't need client instance anymore.

func NewJsonClient added in v0.8.0 

func NewJsonClient(endpoint string, config Config) *Client

NewJsonClient initializes Client which uses JSON-based protocol internally. After client initialized call Client.Connect method. Use Client.NewSubscription to create Subscription objects. The provided endpoint must be a valid URL with ws:// or wss:// scheme – otherwise NewJsonClient will panic.

func NewProtobufClient added in v0.8.0 

func NewProtobufClient(endpoint string, config Config) *Client

NewProtobufClient initializes Client which uses Protobuf-based protocol internally. After client initialized call Client.Connect method. Use Client.NewSubscription to create Subscription objects. The provided endpoint must be a valid URL with ws:// or wss:// scheme – otherwise NewProtobufClient will panic.

func (*Client) Close 

func (c *Client) Close()

Close closes Client and cleanups resources. Client is unusable after this. Use this method if you don't need client anymore, otherwise look at Client.Disconnect.

func (*Client) Connect 

func (c *Client) Connect() error

Connect dials to server and sends connect message. Will return an error if first dial with a server failed. In case of failure client will automatically reconnect. To temporary disconnect from a server call Client.Disconnect.

func (*Client) Disconnect 

func (c *Client) Disconnect() error

Disconnect client from server. It's still possible to connect again later. If you don't need Client anymore – use Client.Close.

func (*Client) GetSubscription added in v0.9.0 

func (c *Client) GetSubscription(channel string) (*Subscription, bool)

GetSubscription allows getting Subscription from the internal client registry.

func (*Client) History added in v0.7.0 

func (c *Client) History(ctx context.Context, channel string, opts ...HistoryOption) (HistoryResult, error)

History for a channel without being subscribed.

func (*Client) NewSubscription added in v0.2.0 

func (c *Client) NewSubscription(channel string, config ...SubscriptionConfig) (*Subscription, error)

NewSubscription allocates new Subscription on a channel. As soon as Subscription successfully created Client keeps reference to it inside internal map registry to manage automatic resubscribe on reconnect. After creating Subscription call its Subscription.Subscribe method to actually start subscribing process. To temporarily unsubscribe call Subscription.Unsubscribe. If you finished with Subscription then you can remove it from the internal registry by calling Client.RemoveSubscription method.

func (*Client) OnConnected added in v0.9.0 

func (c *Client) OnConnected(handler ConnectedHandler)

OnConnected is a function to handle connect event.

func (*Client) OnConnecting added in v0.9.0 

func (c *Client) OnConnecting(handler ConnectingHandler)

OnConnecting is a function to handle connecting event.

func (*Client) OnDisconnected added in v0.9.0 

func (c *Client) OnDisconnected(handler DisconnectHandler)

OnDisconnected is a function to handle moveToDisconnected event.

func (*Client) OnError added in v0.2.0 

func (c *Client) OnError(handler ErrorHandler)

OnError is a function that will receive unhandled errors for logging.

func (*Client) OnJoin added in v0.9.0 

func (c *Client) OnJoin(handler ServerJoinHandler)

OnJoin sets function to handle Join event from server-side subscriptions.

func (*Client) OnLeave added in v0.9.0 

func (c *Client) OnLeave(handler ServerLeaveHandler)

OnLeave sets function to handle Leave event from server-side subscriptions.

func (*Client) OnMessage added in v0.2.0 

func (c *Client) OnMessage(handler MessageHandler)

OnMessage allows processing async message from server to client.

func (*Client) OnPublication added in v0.9.0 

func (c *Client) OnPublication(handler ServerPublicationHandler)

OnPublication sets function to handle Publications from server-side subscriptions.

func (*Client) OnSubscribed added in v0.9.0 

func (c *Client) OnSubscribed(handler ServerSubscribedHandler)

OnSubscribed sets function to handle server-side subscription subscribe events.

func (*Client) OnSubscribing added in v0.9.0 

func (c *Client) OnSubscribing(handler ServerSubscribingHandler)

OnSubscribing sets function to handle server-side subscription subscribing events.

func (*Client) OnUnsubscribed added in v0.9.0 

func (c *Client) OnUnsubscribed(handler ServerUnsubscribedHandler)

OnUnsubscribed sets function to handle unsubscribe from server-side subscriptions.

func (*Client) Presence added in v0.7.0 

func (c *Client) Presence(ctx context.Context, channel string) (PresenceResult, error)

Presence for a channel without being subscribed.

func (*Client) PresenceStats added in v0.7.0 

func (c *Client) PresenceStats(ctx context.Context, channel string) (PresenceStatsResult, error)

PresenceStats for a channel without being subscribed.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, channel string, data []byte) (PublishResult, error)

Publish data into channel.

func (*Client) RPC 

func (c *Client) RPC(ctx context.Context, method string, data []byte) (RPCResult, error)

RPC allows sending data to a server and waiting for a response. RPC handler must be registered on server.

func (*Client) RemoveSubscription added in v0.9.0 

func (c *Client) RemoveSubscription(sub *Subscription) error

RemoveSubscription removes Subscription from the internal client registry. Make sure Subscription is in unsubscribed state before removing it.

func (*Client) Send 

func (c *Client) Send(ctx context.Context, data []byte) error

Send message to server without waiting for response. Message handler must be registered on server.

func (*Client) SetToken 

func (c *Client) SetToken(token string)

SetToken allows updating Client's connection token.

func (*Client) State added in v0.9.0 

func (c *Client) State() State

State returns current Client state. Note that while you are processing this state - Client can move to a new one.

func (*Client) Subscriptions added in v0.9.0 

func (c *Client) Subscriptions() map[string]*Subscription

Subscriptions returns a map with all currently registered client-side subscriptions.

type ClientInfo 

type ClientInfo struct {
	// Client is a client unique id.
	Client string
	// User is an ID of authenticated user. Zero value means anonymous user.
	User string
	// ConnInfo is additional information about connection.
	ConnInfo []byte
	// ChanInfo is additional information about connection in context of
	// channel subscription.
	ChanInfo []byte
}

ClientInfo contains information about client connection.

type Config 

type Config struct {
	// Token for a connection authentication.
	Token string
	// GetToken called by SDK to get or refresh connection token.
	GetToken func(ConnectionTokenEvent) (string, error)
	// Data is an arbitrary data which can be sent to a server in a Connect command.
	// Make sure it's a valid JSON when using JSON protocol client.
	Data []byte
	// CookieJar specifies the cookie jar to send in WebSocket Upgrade request.
	CookieJar http.CookieJar
	// Header specifies custom HTTP Header to send in WebSocket Upgrade request.
	Header http.Header
	// Name allows setting client name. You should only use a limited
	// amount of client names throughout your applications – i.e. don't
	// make it unique per user for example, this name semantically represents
	// an environment from which client connects.
	// Zero value means "go".
	Name string
	// Version allows setting client version. This is an application
	// specific information. By default, no version set.
	Version string
	// Proxy specifies a function to return a proxy for a given Request.
	// If the function returns a non-nil error, the request is aborted with the
	// provided error. If function returns a nil *URL, no proxy is used.
	// If Proxy is nil then http.ProxyFromEnvironment will be used.
	Proxy func(*http.Request) (*url.URL, error)
	// NetDialContext specifies the dial function for creating TCP connections. If
	// NetDialContext is nil, net.DialContext is used.
	NetDialContext func(ctx context.Context, network, addr string) (net.Conn, error)
	// ReadTimeout is how long to wait read operations to complete.
	// Zero value means 5 * time.Second.
	ReadTimeout time.Duration
	// WriteTimeout is Websocket write timeout.
	// Zero value means 1 * time.Second.
	WriteTimeout time.Duration
	// HandshakeTimeout specifies the duration for the handshake to complete.
	// Zero value means 1 * time.Second.
	HandshakeTimeout time.Duration
	// MaxServerPingDelay used to set maximum delay of ping from server.
	// Zero value means 10 * time.Second.
	MaxServerPingDelay time.Duration
	// MinReconnectDelay is the minimum delay between reconnection attempts.
	// This delay is jittered.
	// Zero value means 200 * time.Millisecond.
	MinReconnectDelay time.Duration
	// MaxReconnectDelay is the maximum delay between reconnection attempts.
	// Zero value means 20 * time.Second.
	MaxReconnectDelay time.Duration
	// TLSConfig specifies the TLS configuration to use with tls.Client.
	// If nil, the default configuration is used.
	TLSConfig *tls.Config
	// EnableCompression specifies if the client should attempt to negotiate
	// per message compression (RFC 7692). Setting this value to true does not
	// guarantee that compression will be supported. Currently, only "no context
	// takeover" modes are supported.
	EnableCompression bool
	// LogLevel to use, by default no logs will be exposed by centrifuge-go. Most of the
	// time available protocol callbacks cover all necessary information about client-server
	// communication.
	LogLevel LogLevel
	// LogHandler is a function that will be called for each log entry. Log entries
	// are sent asynchronously and from a separate goroutine by centrifuge-go with an
	// intermediary channel buffer (fixed capacity 256). If your LogHandler is not
	// processing log entries fast enough, centrifuge-go will drop log entries.
	LogHandler func(LogEntry)
}

Config contains various client options.

type ConfigurationError added in v0.10.0 

type ConfigurationError struct {
	Err error
}

func (ConfigurationError) Error added in v0.10.0 

func (c ConfigurationError) Error() string

func (ConfigurationError) Unwrap added in v0.10.4 

func (c ConfigurationError) Unwrap() error

type ConnectError added in v0.9.0 

type ConnectError struct {
	Err error
}

func (ConnectError) Error added in v0.9.0 

func (c ConnectError) Error() string

func (ConnectError) Unwrap added in v0.10.4 

func (c ConnectError) Unwrap() error

type ConnectedEvent added in v0.9.0 

type ConnectedEvent struct {
	ClientID string
	Version  string
	Data     []byte
}

ConnectedEvent is a connected event context passed to OnConnected callback.

type ConnectedHandler added in v0.9.0 

type ConnectedHandler func(ConnectedEvent)

ConnectedHandler is an interface describing how to handle connect event.

type ConnectingEvent added in v0.9.0 

type ConnectingEvent struct {
	Code   uint32
	Reason string
}

ConnectingEvent is a connecting event context passed to OnConnecting callback.

type ConnectingHandler added in v0.9.0 

type ConnectingHandler func(ConnectingEvent)

ConnectingHandler is an interface describing how to handle connecting event.

type ConnectionTokenEvent added in v0.9.0 

type ConnectionTokenEvent struct {
}

ConnectionTokenEvent may contain some useful contextual information in the future. For now, it's empty.

type DeltaError added in v0.13.0 

type DeltaError struct {
	Channel string
	Offset  uint64
	Err     error
}

DeltaError is reported through Client.OnError when a publication's delta can't be applied. The client then disconnects with a bad protocol code and doesn't reconnect: a broken delta chain would give the application wrong data.

func (DeltaError) Error added in v0.13.0 

func (d DeltaError) Error() string

func (DeltaError) Unwrap added in v0.13.0 

func (d DeltaError) Unwrap() error

type DeltaType added in v0.10.7 

type DeltaType string

DeltaType represents type of delta used for Subscription. 

const (
	// DeltaTypeNone means that no delta is used for this subscription.
	DeltaTypeNone DeltaType = ""
	// DeltaTypeFossil means Fossil-based delta.
	DeltaTypeFossil DeltaType = "fossil"
)

type DisconnectHandler 

type DisconnectHandler func(DisconnectedEvent)

DisconnectHandler is an interface describing how to handle moveToDisconnected event.

type DisconnectedEvent added in v0.9.0 

type DisconnectedEvent struct {
	Code   uint32
	Reason string
}

DisconnectedEvent is a disconnected event context passed to OnDisconnected callback.

type Error 

type Error struct {
	Code      uint32
	Message   string
	Temporary bool
}

Error represents protocol-level error.

func (Error) Error added in v0.7.0 

func (e Error) Error() string

type ErrorEvent 

type ErrorEvent struct {
	Error error
}

ErrorEvent is an error event context passed to OnError callback.

type ErrorHandler 

type ErrorHandler func(ErrorEvent)

ErrorHandler is an interface describing how to handle error event.

type FilterNode added in v0.12.0 

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

FilterNode is a node in a publication tags-filter expression tree, used for server-side publication filtering. Build instances with the Filter* helpers and pass the result via SubscriptionConfig.TagsFilter or Subscription.SetTagsFilter.

A filter is either a leaf node (a comparison such as ticker == "AAPL") or a logical node combining child nodes with and/or/not. The server evaluates the filter against each publication's tags and delivers only matching publications. See https://centrifugal.dev/docs/server/publication_filtering.

Publication filtering must be enabled for the namespace on the server (allow_tags_filter) and cannot be combined with delta compression.

func FilterAnd added in v0.12.0 

func FilterAnd(nodes ...*FilterNode) *FilterNode

FilterAnd matches when all nodes match.

func FilterContains added in v0.12.0 

func FilterContains(key, val string) *FilterNode

FilterContains matches when string tag key contains val.

func FilterEndsWith added in v0.12.0 

func FilterEndsWith(key, val string) *FilterNode

FilterEndsWith matches when string tag key ends with val.

func FilterEq added in v0.12.0 

func FilterEq(key, val string) *FilterNode

FilterEq matches when tag key equals val.

func FilterExists added in v0.12.0 

func FilterExists(key string) *FilterNode

FilterExists matches when tag key exists.

func FilterGt added in v0.12.0 

func FilterGt(key, val string) *FilterNode

FilterGt matches when numeric tag key is greater than val.

func FilterGte added in v0.12.0 

func FilterGte(key, val string) *FilterNode

FilterGte matches when numeric tag key is greater than or equal to val.

func FilterIn added in v0.12.0 

func FilterIn(key string, vals ...string) *FilterNode

FilterIn matches when tag key is one of vals.

func FilterLt added in v0.12.0 

func FilterLt(key, val string) *FilterNode

FilterLt matches when numeric tag key is less than val.

func FilterLte added in v0.12.0 

func FilterLte(key, val string) *FilterNode

FilterLte matches when numeric tag key is less than or equal to val.

func FilterNeq added in v0.12.0 

func FilterNeq(key, val string) *FilterNode

FilterNeq matches when tag key does not equal val.

func FilterNin added in v0.12.0 

func FilterNin(key string, vals ...string) *FilterNode

FilterNin matches when tag key is not one of vals.

func FilterNot added in v0.12.0 

func FilterNot(node *FilterNode) *FilterNode

FilterNot inverts node.

func FilterNotExists added in v0.12.0 

func FilterNotExists(key string) *FilterNode

FilterNotExists matches when tag key does not exist.

func FilterOr added in v0.12.0 

func FilterOr(nodes ...*FilterNode) *FilterNode

FilterOr matches when at least one of nodes matches.

func FilterStartsWith added in v0.12.0 

func FilterStartsWith(key, val string) *FilterNode

FilterStartsWith matches when string tag key starts with val.

type HistoryOption added in v0.8.0 

type HistoryOption func(options *HistoryOptions)

func WithHistoryLimit added in v0.8.0 

func WithHistoryLimit(limit int32) HistoryOption

func WithHistoryReverse added in v0.8.1 

func WithHistoryReverse(reverse bool) HistoryOption

func WithHistorySince added in v0.8.0 

func WithHistorySince(sp *StreamPosition) HistoryOption

type HistoryOptions added in v0.8.0 

type HistoryOptions struct {
	Limit   int32
	Since   *StreamPosition
	Reverse bool
}

type HistoryResult added in v0.6.0 

type HistoryResult struct {
	Publications []Publication
	Offset       uint64
	Epoch        string
}

HistoryResult contains the result of history op.

type JoinEvent 

type JoinEvent struct {
	ClientInfo
}

JoinEvent has info about user who joined channel.

type JoinHandler 

type JoinHandler func(JoinEvent)

JoinHandler is a function to handle join messages.

type LeaveEvent 

type LeaveEvent struct {
	ClientInfo
}

LeaveEvent has info about user who left channel.

type LeaveHandler 

type LeaveHandler func(LeaveEvent)

LeaveHandler is a function to handle leave messages.

type LogEntry added in v0.10.7 

type LogEntry struct {
	Level   LogLevel
	Message string
	Fields  map[string]string
}

LogEntry describes a log entry.

type LogHandler added in v0.10.7 

type LogHandler func(LogEntry)

LogHandler handles log entries - i.e. writes into correct destination if necessary.

type LogLevel added in v0.10.7 

type LogLevel int

LogLevel describes the chosen log level. 

const (
	// LogLevelNone means no logging.
	LogLevelNone LogLevel = iota
	// LogLevelTrace turns on trace logs - should only be used during development. This
	// log level shows all client-server communication (including token content) and
	// affects performance significantly.
	LogLevelTrace
	// LogLevelDebug turns on debug logs - it's generally too much for production in normal
	// conditions but can help when developing and investigating problems in production.
	LogLevelDebug
)

func (LogLevel) String added in v0.10.7 

func (l LogLevel) String() string

type MessageEvent 

type MessageEvent struct {
	Data []byte
}

MessageEvent is an event for async message from server to client.

type MessageHandler 

type MessageHandler func(MessageEvent)

MessageHandler is an interface describing how to handle async message from server.

type PresenceResult added in v0.6.0 

type PresenceResult struct {
	Clients map[string]ClientInfo
}

PresenceResult contains the result of presence op.

type PresenceStats 

type PresenceStats struct {
	NumClients int
	NumUsers   int
}

PresenceStats represents short presence information.

type PresenceStatsResult added in v0.6.0 

type PresenceStatsResult struct {
	PresenceStats
}

PresenceStatsResult wraps presence stats.

type Publication 

type Publication struct {
	// Offset is an incremental position number inside history stream.
	// Zero value means that channel does not maintain Publication stream.
	Offset uint64
	// Data published to channel.
	Data []byte
	// Info is optional information about client connection published
	// this data to channel.
	Info *ClientInfo
	// Tags contain custom key-value pairs attached to Publication.
	Tags map[string]string
}

Publication is a data sent to channel.

type PublicationEvent added in v0.9.0 

type PublicationEvent struct {
	Publication
}

PublicationEvent has info about received channel Publication.

type PublicationHandler added in v0.9.0 

type PublicationHandler func(PublicationEvent)

PublicationHandler is a function to handle messages published in channels.

type PublishResult added in v0.6.0 

type PublishResult struct{}

PublishResult contains the result of publish.

type RPCResult added in v0.6.0 

type RPCResult struct {
	Data []byte
}

RPCResult contains data returned from server as RPC result.

type RefreshError added in v0.9.0 

type RefreshError struct {
	Err error
}

func (RefreshError) Error added in v0.9.0 

func (r RefreshError) Error() string

func (RefreshError) Unwrap added in v0.10.4 

func (r RefreshError) Unwrap() error

type ServerJoinEvent added in v0.6.0 

type ServerJoinEvent struct {
	Channel string
	ClientInfo
}

ServerJoinEvent has info about user who joined channel.

type ServerJoinHandler added in v0.6.0 

type ServerJoinHandler func(ServerJoinEvent)

ServerJoinHandler is an interface describing how to handle Join events from server-side subscriptions.

type ServerLeaveEvent added in v0.6.0 

type ServerLeaveEvent struct {
	Channel string
	ClientInfo
}

ServerLeaveEvent has info about user who left channel.

type ServerLeaveHandler added in v0.6.0 

type ServerLeaveHandler func(ServerLeaveEvent)

ServerLeaveHandler is an interface describing how to handle Leave events from server-side subscriptions.

type ServerPublicationEvent added in v0.9.0 

type ServerPublicationEvent struct {
	Channel string
	Publication
}

ServerPublicationEvent has info about received channel Publication.

type ServerPublicationHandler added in v0.9.0 

type ServerPublicationHandler func(ServerPublicationEvent)

ServerPublicationHandler is an interface describing how to handle Publication from server-side subscriptions.

type ServerSubscribedEvent added in v0.9.0 

type ServerSubscribedEvent struct {
	Channel        string
	WasRecovering  bool
	Recovered      bool
	Recoverable    bool
	Positioned     bool
	StreamPosition *StreamPosition
	Data           []byte
}

ServerSubscribedEvent has info about server-side subscription.

type ServerSubscribedHandler added in v0.9.0 

type ServerSubscribedHandler func(ServerSubscribedEvent)

ServerSubscribedHandler is an interface describing how to handle subscribe events from server-side subscriptions.

type ServerSubscribingEvent added in v0.9.0 

type ServerSubscribingEvent struct {
	Channel string
}

ServerSubscribingEvent is an event passed to subscribing event handler.

type ServerSubscribingHandler added in v0.9.0 

type ServerSubscribingHandler func(ServerSubscribingEvent)

ServerSubscribingHandler is an interface describing how to handle subscribing events for server-side subscriptions.

type ServerUnsubscribedEvent added in v0.9.0 

type ServerUnsubscribedEvent struct {
	Channel string
}

ServerUnsubscribedEvent is an event passed to unsubscribe event handler.

type ServerUnsubscribedHandler added in v0.9.0 

type ServerUnsubscribedHandler func(ServerUnsubscribedEvent)

ServerUnsubscribedHandler is an interface describing how to handle unsubscribe events from server-side subscriptions.

type State added in v0.9.0 

type State string

State of client connection. 

const (
	StateDisconnected State = "disconnected"
	StateConnecting   State = "connecting"
	StateConnected    State = "connected"
	StateClosed       State = "closed"
)

Describe client connection states.

type StreamPosition added in v0.8.0 

type StreamPosition struct {
	Offset uint64
	Epoch  string
}

type SubState added in v0.9.0 

type SubState string

SubState represents state of Subscription. 

const (
	SubStateUnsubscribed SubState = "unsubscribed"
	SubStateSubscribing  SubState = "subscribing"
	SubStateSubscribed   SubState = "subscribed"
)

Different states of Subscription.

type SubscribedEvent added in v0.9.0 

type SubscribedEvent struct {
	Positioned     bool
	Recoverable    bool
	StreamPosition *StreamPosition
	WasRecovering  bool
	Recovered      bool
	Data           []byte
}

SubscribedEvent is an event context passed to subscribe success callback.

type SubscribedHandler added in v0.9.0 

type SubscribedHandler func(SubscribedEvent)

SubscribedHandler is a function to handle subscribe success event.

type SubscribingEvent added in v0.9.0 

type SubscribingEvent struct {
	Code   uint32
	Reason string
}

SubscribingEvent is an event passed to subscribing event handler.

type SubscribingHandler added in v0.9.0 

type SubscribingHandler func(SubscribingEvent)

SubscribingHandler is a function to handle subscribe success event.

type Subscription 

type Subscription struct {

	// Channel for a subscription.
	Channel string
	// contains filtered or unexported fields
}

Subscription represents client subscription to channel. DO NOT initialize this struct directly, instead use Client.NewSubscription method to create channel subscriptions.

func (*Subscription) History 

func (s *Subscription) History(ctx context.Context, opts ...HistoryOption) (HistoryResult, error)

History allows extracting channel history. By default, it returns current stream top position without publications. Use WithHistoryLimit with a value > 0 to make this func to return publications.

func (*Subscription) OnError added in v0.9.0 

func (s *Subscription) OnError(handler SubscriptionErrorHandler)

OnError allows setting SubscriptionErrorHandler to SubEventHandler.

func (*Subscription) OnJoin added in v0.2.0 

func (s *Subscription) OnJoin(handler JoinHandler)

OnJoin allows setting JoinHandler to SubEventHandler.

func (*Subscription) OnLeave added in v0.2.0 

func (s *Subscription) OnLeave(handler LeaveHandler)

OnLeave allows setting LeaveHandler to SubEventHandler.

func (*Subscription) OnPublication added in v0.9.0 

func (s *Subscription) OnPublication(handler PublicationHandler)

OnPublication allows setting PublicationHandler to SubEventHandler.

func (*Subscription) OnSubscribed added in v0.9.0 

func (s *Subscription) OnSubscribed(handler SubscribedHandler)

OnSubscribed allows setting SubscribedHandler to SubEventHandler.

func (*Subscription) OnSubscribing added in v0.9.0 

func (s *Subscription) OnSubscribing(handler SubscribingHandler)

OnSubscribing allows setting SubscribingHandler to SubEventHandler.

func (*Subscription) OnUnsubscribed added in v0.9.0 

func (s *Subscription) OnUnsubscribed(handler UnsubscribedHandler)

OnUnsubscribed allows setting UnsubscribedHandler to SubEventHandler.

func (*Subscription) Presence 

func (s *Subscription) Presence(ctx context.Context) (PresenceResult, error)

Presence allows extracting channel presence.

func (*Subscription) PresenceStats 

func (s *Subscription) PresenceStats(ctx context.Context) (PresenceStatsResult, error)

PresenceStats allows extracting channel presence stats.

func (*Subscription) Publish 

func (s *Subscription) Publish(ctx context.Context, data []byte) (PublishResult, error)

Publish allows publishing data to the subscription channel.

func (*Subscription) SetTagsFilter added in v0.12.0 

func (s *Subscription) SetTagsFilter(tagsFilter *FilterNode) error

SetTagsFilter sets the server-side publication tags filter. It is applied on the next subscribe attempt, not the current one. Pass nil to clear it. Cannot be combined with delta compression. Build with the Filter* helpers.

func (*Subscription) State added in v0.9.0 

func (s *Subscription) State() SubState

func (*Subscription) Subscribe 

func (s *Subscription) Subscribe() error

Subscribe allows initiating subscription process.

func (*Subscription) Unsubscribe 

func (s *Subscription) Unsubscribe() error

Unsubscribe allows unsubscribing from channel.

type SubscriptionConfig added in v0.9.0 

type SubscriptionConfig struct {
	// Data is an arbitrary data to pass to a server in each subscribe request.
	Data []byte
	// Token for Subscription.
	Token string
	// GetToken called to get or refresh private channel subscription token.
	GetToken func(SubscriptionTokenEvent) (string, error)
	// Positioned flag asks server to make Subscription positioned. Only makes sense
	// in channels with history stream on.
	Positioned bool
	// Recoverable flag asks server to make Subscription recoverable. Only makes sense
	// in channels with history stream on.
	Recoverable bool
	// JoinLeave flag asks server to push join/leave messages.
	JoinLeave bool
	// Delta allows to specify delta type for the subscription. By default, no delta is used.
	Delta DeltaType
	// TagsFilter sets a server-side publication filter based on publication tags.
	// When set, the server delivers only publications whose tags match the
	// filter. Must be enabled for the namespace on the server (allow_tags_filter)
	// and cannot be combined with Delta. Build with the Filter* helpers.
	TagsFilter *FilterNode
	// MinResubscribeDelay is the minimum delay between resubscription attempts.
	// This delay is jittered.
	// Zero value means 200 * time.Millisecond.
	MinResubscribeDelay time.Duration
	// MaxResubscribeDelay is the maximum delay between resubscription attempts.
	// Zero value means 20 * time.Second.
	MaxResubscribeDelay time.Duration
	// GetState called to load the app's current state and stream position.
	// Requires Centrifugo >= 6.8.0.
	//
	// The SDK calls this:
	// - On initial subscribe (no saved position)
	// - On reconnect when recovery fails (server returns error 112 —
	//   unrecoverable position)
	//
	// NOT called on reconnects where the server successfully recovers missed
	// publications — in that case the recovered publications arrive as events
	// and GetState is skipped.
	//
	// The app should load its data from its own source of truth (database,
	// API), render it, and return the stream position. The SDK subscribes with
	// recovery from the returned position, so any publications between the
	// state read and the subscribe are delivered as publication events.
	//
	// IMPORTANT: inside GetState, read the stream position FIRST, then read
	// your data. This ensures the position is a lower bound — any data loaded
	// after the position read is guaranteed to be included. The reverse order
	// can produce gaps.
	//
	// Recovered publications may overlap with data already loaded in GetState.
	// This works correctly when updates are idempotent (applying the same
	// update twice produces the same result). For non-idempotent updates,
	// deduplicate by publication offset.
	//
	// On error, the SDK emits an error event with SubscriptionGetStateError
	// and retries with backoff.
	GetState func(SubscriptionGetStateEvent) (StreamPosition, error)
}

SubscriptionConfig allows setting Subscription options.

type SubscriptionErrorEvent added in v0.9.0 

type SubscriptionErrorEvent struct {
	Error error
}

SubscriptionErrorEvent is a subscribe error event context passed to event callback.

type SubscriptionErrorHandler added in v0.9.0 

type SubscriptionErrorHandler func(SubscriptionErrorEvent)

SubscriptionErrorHandler is a function to handle subscribe error event.

type SubscriptionGetStateError added in v0.12.0 

type SubscriptionGetStateError struct {
	Err error
}

func (SubscriptionGetStateError) Error added in v0.12.0 

func (s SubscriptionGetStateError) Error() string

func (SubscriptionGetStateError) Unwrap added in v0.12.0 

func (s SubscriptionGetStateError) Unwrap() error

type SubscriptionGetStateEvent added in v0.12.0 

type SubscriptionGetStateEvent struct {
	Channel string
}

SubscriptionGetStateEvent is passed to SubscriptionConfig.GetState callback.

type SubscriptionRefreshError added in v0.9.0 

type SubscriptionRefreshError struct {
	Err error
}

func (SubscriptionRefreshError) Error added in v0.9.0 

func (s SubscriptionRefreshError) Error() string

func (SubscriptionRefreshError) Unwrap added in v0.10.4 

func (s SubscriptionRefreshError) Unwrap() error

type SubscriptionSubscribeError added in v0.9.0 

type SubscriptionSubscribeError struct {
	Err error
}

func (SubscriptionSubscribeError) Error added in v0.9.0 

func (s SubscriptionSubscribeError) Error() string

func (SubscriptionSubscribeError) Unwrap added in v0.10.4 

func (s SubscriptionSubscribeError) Unwrap() error

type SubscriptionTokenEvent added in v0.9.0 

type SubscriptionTokenEvent struct {
	Channel string
}

SubscriptionTokenEvent contains info required to get subscription token when client wants to subscribe on private channel.

type TransportError added in v0.9.0 

type TransportError struct {
	Err error
}

func (TransportError) Error added in v0.9.0 

func (t TransportError) Error() string

func (TransportError) Unwrap added in v0.10.4 

func (t TransportError) Unwrap() error

type UnsubscribeResult added in v0.6.0 

type UnsubscribeResult struct{}

type UnsubscribedEvent added in v0.9.0 

type UnsubscribedEvent struct {
	Code   uint32
	Reason string
}

UnsubscribedEvent is an event passed to unsubscribe event handler.

type UnsubscribedHandler added in v0.9.0 

type UnsubscribedHandler func(UnsubscribedEvent)

UnsubscribedHandler is a function to handle unsubscribe event.

Source Files

View all Source files

Directories

Path Synopsis
examples module

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.