centrifuge

package module
v0.0.0-...-b165c23 Latest Latest
Warning

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

 Go to latest Published: Nov 16, 2018 License: MIT Imports: 32 Imported by: 0

README

Build Status GoDoc

Work in progress. This library has no stable release yet so API can be changed.

Centrifuge library represents real-time core for Centrifugo server. It's also aimed to be a general purpose real-time messaging library for Go programming language.

Message transports:

  • Websocket transport with JSON or binary Protobuf protocol
  • SockJS polyfill library support (JSON only)

Features:

  • Fast and optimized for low-latency communication with thousands of client connections
  • Scaling to many nodes with Redis PUB/SUB, built-in Redis sharding, Sentinel for HA
  • Bidirectional asynchronous message communication, RPC calls
  • Channel (room) concept to broadcast message to all channel subscribers
  • Presence information for channels (show all active clients in channel)
  • History information for channels (last messages published into channel)
  • Join/leave events for channels (aka client goes online/offline)
  • Message recovery mechanism for channels to survive short network disconnects
  • MIT license

Clients (also work in progress but with most features already supported):

Godoc and examples

Installation

To install globally into $GOPATH use:

go get -u github.com/centrifugal/centrifuge

But recommended way is using tools like dep or go mod to add this library as dependency to your project.

Quick example

Let's take a look on how to build the simplest real-time chat ever with Centrifuge library. Clients will be able to open page in browser, connect to server over Websocket, send message into channel and this message will be instantly delivered to all active channel subscribers. On server side we will accept all connections and will work as simple PUB/SUB proxy without worrying too much about permissions. In this example we will use Centrifuge Javascript client on frontend.

Create file main.go with the following code:

package main

import (
	"context"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	// Import this library.
	"github.com/centrifugal/centrifuge"
)

func handleLog(e centrifuge.LogEntry) {
	log.Printf("%s: %v", e.Message, e.Fields)
}

// Wait until program interrupted. When interrupted gracefully shutdown Node.
func waitExitSignal(n *centrifuge.Node) {
	sigs := make(chan os.Signal, 1)
	done := make(chan bool, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigs
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()
		n.Shutdown(ctx)
		done <- true
	}()
	<-done
}

func main() {
	// We use default config here as starting point. Default config contains
	// reasonable values for available options.
	cfg := centrifuge.DefaultConfig
	// In this example we want client to do all possible actions with server
	// without any authentication and authorization. Insecure flag DISABLES
	// many security related checks in library. This is only to make example
	// short. In real app you most probably want authenticate and authorize
	// access to server. See godoc and examples in repo for more details.
	cfg.ClientInsecure = true
	// By default clients can not publish messages into channels. Setting this
	// option to true we allow them to publish.
	cfg.Publish = true

	// Node is the core object in Centrifuge library responsible for many useful
	// things. Here we initialize new Node instance and pass config to it.
	node, _ := centrifuge.New(cfg)

	// On().Connect() method is a point where you create a binding between
	// Centrifuge and your app business logic. Callback function you pass
	// to On().Connect will be called every time new connection established
	// with server. Inside this callback function you can set various event
	// handlers for incoming client connection.
	node.On().Connect(func(ctx context.Context, client *centrifuge.Client, e centrifuge.ConnectEvent) centrifuge.ConnectReply {
		// Set Subscribe Handler to react on every channel subscribtion attempt
		// initiated by client. Here you can theoretically return an error or
		// disconnect client from server if needed. But now we just accept
		// all subscriptions.
		client.On().Subscribe(func(e centrifuge.SubscribeEvent) centrifuge.SubscribeReply {
			log.Printf("client subscribes on channel %s", e.Channel)
			return centrifuge.SubscribeReply{}
		})

		// Set Publish Handler to react on every channel Publication sent by client.
		// Inside this method you can validate client permissions to publish into
		// channel. But in our simple chat app we allow everyone to publish into
		// any channel.
		client.On().Publish(func(e centrifuge.PublishEvent) centrifuge.PublishReply {
			log.Printf("client publishes into channel %s: %s", e.Channel, string(e.Data))
			return centrifuge.PublishReply{}
		})

		// Set Disconnect Handler to react on client disconnect events.
		client.On().Disconnect(func(e centrifuge.DisconnectEvent) centrifuge.DisconnectReply {
			log.Printf("client disconnected")
			return centrifuge.DisconnectReply{}
		})

		// In our example transport will always be Websocket but it can also be SockJS.
		transportName := client.Transport().Name()
		// In our example clients connect with JSON protocol but it can also be Protobuf.
		transportEncoding := client.Transport().Encoding()

		log.Printf("client connected via %s (%s)", transportName, transportEncoding)
		return centrifuge.ConnectReply{}
	})

	// Centrifuge library exposes logs with different log level. In your app
	// you can set special function to handle these log entries in a way you want.
	node.SetLogHandler(centrifuge.LogLevelDebug, handleLog)

	// Run node will start node's underlying Engine, launch several
	// internal goroutines.
	if err := node.Run(); err != nil {
		panic(err)
	}

	// Configure http routes.

	// The first route is for handling Websocket connections.
	http.Handle("/connection/websocket", centrifuge.NewWebsocketHandler(node, centrifuge.WebsocketConfig{}))

	// The second route is for serving index.html file.
	http.Handle("/", http.FileServer(http.Dir("./")))

	// Start HTTP server.
	go func() {
		if err := http.ListenAndServe(":8000", nil); err != nil {
			panic(err)
		}
	}()

	// Run program until interrupted.
	waitExitSignal(node)
}

Also create file index.html near main.go with content:

<!DOCTYPE html>
<html>
    <head>
        <meta charset="utf-8">
        <!-- Note that we use client from c2 branch of centrifuge-js because its not released yet -->
        <!-- TODO: use client from master branch after client release -->
        <script type="text/javascript" src="https://rawgit.com/centrifugal/centrifuge-js/c2/dist/centrifuge.min.js"></script>
    </head>
    <body>
        <input type="text" id="input" />
        <script type="text/javascript">
            // Create Centrifuge object with Websocket endpoint address set in main.go
            var centrifuge = new Centrifuge('ws://localhost:8000/connection/websocket');
            function drawText(text) {
                var div = document.createElement('div');
                div.innerHTML = text;
                document.body.appendChild(div);
            }
            centrifuge.on('connect', function(ctx){
                drawText('Connected over ' + ctx.transport + '<br>');
            });
            centrifuge.on('disconnect', function(ctx){
                drawText('Disconnected: ' + ctx.reason + '<br>');
            });
            var sub = centrifuge.subscribe("chat", function(message) {
                drawText(JSON.stringify(message) + '<br>');
            })
            var input = document.getElementById("input");
            input.addEventListener('keyup', function(e) {
                if (e.keyCode == 13) { // ENTER key pressed
                    sub.publish(this.value);
                    input.value = '';
                }
            });
            // After setting event handlers – initiate actual connection with server.
            centrifuge.connect();
        </script>
    </body>
</html>

Then run Go program as usual:

go run main.go

Open several browser tabs with http://localhost:8000 and see chat in action.

This example is only the top of an iceberg. But it should give you an insight on library API.

Keep in mind that Centrifuge library is not a framework to build chat apps. It's a general purpose real-time transport for your messages with some helpful primitives. You can build many kinds of real-time apps on top of this library including chats but depending on application you may need to write business logic yourself.

For contributors

Currently library uses dep to manage dependencies. This is how you can clone library and install all required dependencies locally:

mkdir -p $GOPATH/src/github.com/centrifugal
git clone https://github.com/centrifugal/centrifuge.git $GOPATH/src/github.com/centrifugal/centrifuge
cd $GOPATH/src/github.com/centrifugal/centrifuge
dep ensure

Another way is using go get but all dependencies will be downloaded into your global $GOPATH in this case:

go get -u github.com/centrifugal/centrifuge

Documentation

Overview

Package centrifuge is a real-time messaging library that abstracts several bidirectional transports (Websocket, SockJS) and provides primitives to build real-time applications with Go. It's also used as core of Centrifugo server.

The API of this library is almost all goroutine-safe except cases where one-time operations like setting callback handlers performed.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// DisconnectNormal is clean disconnect when client cleanly closes connection.
	DisconnectNormal = &Disconnect{
		Reason:    "",
		Reconnect: true,
	}
	// DisconnectShutdown sent when node is going to shut down.
	DisconnectShutdown = &Disconnect{
		Reason:    "shutdown",
		Reconnect: true,
	}
	// DisconnectInvalidToken sent when client came with invalid token.
	DisconnectInvalidToken = &Disconnect{
		Reason:    "invalid token",
		Reconnect: false,
	}
	// DisconnectBadRequest sent when client uses malformed protocol
	// frames or wrong order of commands.
	DisconnectBadRequest = &Disconnect{
		Reason:    "bad request",
		Reconnect: false,
	}
	// DisconnectServerError sent when internal error occurred on server.
	DisconnectServerError = &Disconnect{
		Reason:    "internal server error",
		Reconnect: true,
	}
	// DisconnectExpired sent when client connection expired.
	DisconnectExpired = &Disconnect{
		Reason:    "expired",
		Reconnect: true,
	}
	// DisconnectSubExpired sent when client subscription expired.
	DisconnectSubExpired = &Disconnect{
		Reason:    "subscription expired",
		Reconnect: true,
	}
	// DisconnectStale sent to close connection that did not become
	// authenticated in configured interval after dialing.
	DisconnectStale = &Disconnect{
		Reason:    "stale",
		Reconnect: false,
	}
	// DisconnectSlow sent when client can't read messages fast enough.
	DisconnectSlow = &Disconnect{
		Reason:    "slow",
		Reconnect: true,
	}
	// DisconnectWriteError sent when an error occurred while writing to
	// client connection.
	DisconnectWriteError = &Disconnect{
		Reason:    "write error",
		Reconnect: true,
	}
)

Some predefined disconnect structures used by library internally. Though it's always possible to create Disconnect with any field values on the fly.

View Source 
var (
	// ErrorInternal means server error, if returned this is a signal
	// that something went wrong with server itself and client most probably
	// not guilty.
	ErrorInternal = &Error{
		Code:    100,
		Message: "internal server error",
	}
	// ErrUnauthorized says that request is unauthorized.
	ErrorUnauthorized = &Error{
		Code:    101,
		Message: "unauthorized",
	}
	// ErrorNamespaceNotFound means that namespace in channel name does not exist.
	ErrorNamespaceNotFound = &Error{
		Code:    102,
		Message: "namespace not found",
	}
	// ErrorPermissionDenied means that access to resource not allowed.
	ErrorPermissionDenied = &Error{
		Code:    103,
		Message: "permission denied",
	}
	// ErrorMethodNotFound means that method sent in command does not exist.
	ErrorMethodNotFound = &Error{
		Code:    104,
		Message: "method not found",
	}
	// ErrorAlreadySubscribed returned when client wants to subscribe on channel
	// it already subscribed to.
	ErrorAlreadySubscribed = &Error{
		Code:    105,
		Message: "already subscribed",
	}
	// ErrorLimitExceeded says that some sort of limit exceeded, server logs should
	// give more detailed information.
	ErrorLimitExceeded = &Error{
		Code:    106,
		Message: "limit exceeded",
	}
	// ErrorBadRequest says that server can not process received
	// data because it is malformed.
	ErrorBadRequest = &Error{
		Code:    107,
		Message: "bad request",
	}
	// ErrorNotAvailable means that resource is not enabled.
	ErrorNotAvailable = &Error{
		Code:    108,
		Message: "not available",
	}
	// ErrorTokenExpired ...
	ErrorTokenExpired = &Error{
		Code:    109,
		Message: "token expired",
	}
	// ErrorExpired ...
	ErrorExpired = &Error{
		Code:    110,
		Message: "expired",
	}
)

Here we define well-known errors that can be used in client protocol replies. Library user can define own application specific errors. When define new custom error it is recommended to use error codes > 1000 assuming that codes in interval 0-999 reserved by Centrifuge.

View Source 
var DefaultConfig = Config{
	Name: "centrifuge",

	ChannelMaxLength:         255,
	ChannelPrivatePrefix:     "$",
	ChannelNamespaceBoundary: ":",
	ChannelUserBoundary:      "#",
	ChannelUserSeparator:     ",",

	ClientInsecure:               false,
	ClientAnonymous:              false,
	ClientPresencePingInterval:   25 * time.Second,
	ClientPresenceExpireInterval: 60 * time.Second,
	ClientMessageWriteTimeout:    0,
	ClientPingInterval:           25 * time.Second,
	ClientExpiredCloseDelay:      25 * time.Second,
	ClientExpiredSubCloseDelay:   25 * time.Second,
	ClientStaleCloseDelay:        25 * time.Second,
	ClientRequestMaxSize:         65536,
	ClientQueueMaxSize:           10485760,
	ClientChannelLimit:           128,
}

DefaultConfig is Config initialized with default values for all fields.

View Source 
var (
	// ErrNoChannelOptions returned when operation can't be performed because no
	// appropriate channel options were found for channel.
	ErrNoChannelOptions = errors.New("no channel options found")
)
View Source 
var LogStringToLevel = map[string]LogLevel{
	"debug": LogLevelDebug,
	"info":  LogLevelInfo,
	"error": LogLevelError,
	"none":  LogLevelNone,
}

LogStringToLevel matches level string to LogLevel.

Functions

func LogLevelToString 

func LogLevelToString(l LogLevel) string

LogLevelToString transforms Level to its string representation.

func SetCredentials 

func SetCredentials(ctx context.Context, creds *Credentials) context.Context

SetCredentials allows to set connection Credentials to context.

Types

type ChannelContext 

type ChannelContext struct {
	Info proto.Raw
	// contains filtered or unexported fields
}

ChannelContext contains extra context for channel connection subscribed to.

type ChannelNamespace 

type ChannelNamespace struct {
	// Name is a unique namespace name.
	Name string `json:"name"`

	// Options for namespace determine channel options for channels
	// belonging to this namespace.
	ChannelOptions `mapstructure:",squash"`
}

ChannelNamespace allows to create channels with different channel options.

type ChannelOptions 

type ChannelOptions struct {
	// Publish enables possibility for clients to publish messages into channels.
	// Once enabled client can publish into channel and that publication will be
	// broadcasted to all current channel subscribers. You can control publishing
	// on server-side setting On().Publish callback to client connection.
	Publish bool `json:"publish"`

	// SubscribeToPublish turns on an automatic check that client subscribed
	// on channel before allow it to publish into that channel.
	SubscribeToPublish bool `mapstructure:"subscribe_to_publish" json:"subscribe_to_publish"`

	// Anonymous enables anonymous access (with empty user ID) to channel.
	// In most situations your application works with authenticated users so
	// every user has its own unique user ID. But if you provide real-time
	// features for public access you may need unauthenticated access to channels.
	// Turn on this option and use empty string as user ID.
	Anonymous bool `json:"anonymous"`

	// JoinLeave turns on join/leave messages for channels.
	// When client subscribes on channel join message sent to all
	// clients in this channel. When client leaves channel (unsubscribes)
	// leave message sent. This option does not fit well for channels with
	// many subscribers because every subscribe/unsubscribe event results
	// into join/leave event broadcast to all other active subscribers.
	JoinLeave bool `mapstructure:"join_leave" json:"join_leave"`

	// Presence turns on presence information for channels.
	// Presence is a structure with clients currently subscribed on channel.
	Presence bool `json:"presence"`

	// HistorySize determines max amount of history messages for channel,
	// 0 means no history for channel. Centrifugo history has auxiliary
	// role – it can not replace your backend persistent storage.
	HistorySize int `mapstructure:"history_size" json:"history_size"`

	// HistoryLifetime determines time in seconds until expiration for
	// history messages. As Centrifuge-based server keeps history in memory
	// (for example in process memory or in Redis process memory) it's
	// important to remove old messages to prevent infinite memory grows.
	HistoryLifetime int `mapstructure:"history_lifetime" json:"history_lifetime"`

	// Recover enables recover mechanism for channels. This means that
	// server will try to recover missed messages for resubscribing
	// client. This option uses publications from history and must be used
	// with reasonable HistorySize and HistoryLifetime configuration.
	HistoryRecover bool `mapstructure:"history_recover" json:"history_recover"`
}

ChannelOptions represent channel specific configuration for namespace or global channel options if set on top level of configuration.

type Client 

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

Client represents client connection to server.

func (*Client) Channels 

func (c *Client) Channels() map[string]ChannelContext

Channels returns a map of channels client connection currently subscribed to.

func (*Client) Close 

func (c *Client) Close(disconnect *Disconnect) error

Close closes client connection.

func (*Client) ID 

func (c *Client) ID() string

ID returns unique client connection id.

func (*Client) On 

func (c *Client) On() *ClientEventHub

On returns ClientEventHub to set various event handlers to client.

func (*Client) Send 

func (c *Client) Send(data Raw) error

Send data to client connection asynchronously.

func (*Client) Transport 

func (c *Client) Transport() Transport

Transport returns transport used by client connection.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(ch string, resubscribe bool) error

Unsubscribe allows to unsubscribe client from channel.

func (*Client) UserID 

func (c *Client) UserID() string

UserID returns user ID associated with client connection.

type ClientEventHub 

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

ClientEventHub allows to deal with client event handlers. All its methods are not goroutine-safe and supposed to be called once on client connect.

func (*ClientEventHub) Disconnect 

func (c *ClientEventHub) Disconnect(h DisconnectHandler)

Disconnect allows to set DisconnectHandler. DisconnectHandler called when client disconnected.

func (*ClientEventHub) Message 

func (c *ClientEventHub) Message(h MessageHandler)

Message allows to set MessageHandler. MessageHandler called when client sent asynchronous message.

func (*ClientEventHub) Publish 

func (c *ClientEventHub) Publish(h PublishHandler)

Publish allows to set PublishHandler. PublishHandler called when client publishes message into channel.

func (*ClientEventHub) RPC 

func (c *ClientEventHub) RPC(h RPCHandler)

RPC allows to set RPCHandler. RPCHandler will be executed on every incoming RPC call.

func (*ClientEventHub) Refresh 

func (c *ClientEventHub) Refresh(h RefreshHandler)

Refresh allows to set RefreshHandler. RefreshHandler called when it's time to refresh client connection credentials.

func (*ClientEventHub) SubRefresh 

func (c *ClientEventHub) SubRefresh(h SubRefreshHandler)

SubRefresh allows to set SubRefreshHandler. SubRefreshHandler called when it's time to refresh client subscription.

func (*ClientEventHub) Subscribe 

func (c *ClientEventHub) Subscribe(h SubscribeHandler)

Subscribe allows to set SubscribeHandler. SubscribeHandler called when client subscribes on channel.

func (*ClientEventHub) Unsubscribe 

func (c *ClientEventHub) Unsubscribe(h UnsubscribeHandler)

Unsubscribe allows to set UnsubscribeHandler. UnsubscribeHandler called when client unsubscribes from channel.

type ClientInfo 

type ClientInfo = proto.ClientInfo

ClientInfo is short information about client connection.

type Config 

type Config struct {
	// Version of server – will be sent to client on connection establishement
	// phase in response to connect request.
	Version string
	// Name of this server node - must be unique, used as human readable
	// and meaningful node identificator.
	Name string
	// Secret is a secret key used to generate connection and subscription tokens.
	Secret string
	// ChannelOptions embedded.
	ChannelOptions
	// Namespaces – list of namespaces for custom channel options.
	Namespaces []ChannelNamespace
	// ClientInsecure turns on insecure mode for client connections - when it's
	// turned on then no authentication required at all when connecting to Centrifugo,
	// anonymous access and publish allowed for all channels, no connection expire
	// performed. This can be suitable for demonstration or personal usage.
	ClientInsecure bool
	// ClientAnonymous when set to true, allows connect requests without specifying
	// a token or setting Credentials in authentication middleware. The resulting
	// user will have empty string for user ID, meaning user can only subscribe
	// to anonymous channels.
	ClientAnonymous bool
	// ClientPresencePingInterval is an interval how often connected clients
	// must update presence info.
	ClientPresencePingInterval time.Duration
	// ClientPresenceExpireInterval is an interval how long to consider
	// presence info valid after receiving presence ping.
	ClientPresenceExpireInterval time.Duration
	// ClientPingInterval sets interval server will send ping messages to clients.
	ClientPingInterval time.Duration
	// ClientExpiredCloseDelay is an extra time given to client to
	// refresh its connection in the end of connection lifetime.
	ClientExpiredCloseDelay time.Duration
	// ClientExpiredSubCloseDelay is an extra time given to client to
	// refresh its expiring subscription in the end of subscription lifetime.
	ClientExpiredSubCloseDelay time.Duration
	// ClientStaleCloseDelay is a timeout after which connection will be
	// closed if still not authenticated (i.e. no valid connect command
	// received yet).
	ClientStaleCloseDelay time.Duration
	// ClientMessageWriteTimeout is maximum time of write message operation.
	// Slow client will be disconnected. By default we don't use this option (i.e. it's 0)
	// and slow client connections will be closed when there queue size exceeds
	// ClientQueueMaxSize. In case of SockJS transport we don't have control over it so
	// it only affects raw websocket.
	ClientMessageWriteTimeout time.Duration
	// ClientRequestMaxSize sets maximum size in bytes of allowed client request.
	ClientRequestMaxSize int
	// ClientQueueMaxSize is a maximum size of client's message queue in bytes.
	// After this queue size exceeded Centrifugo closes client's connection.
	ClientQueueMaxSize int
	// ClientChannelLimit sets upper limit of channels each client can subscribe to.
	ClientChannelLimit int
	// ClientUserConnectionLimit limits number of client connections from user with the
	// same ID. 0 - unlimited.
	ClientUserConnectionLimit int
	// ChannelPrivatePrefix is a prefix in channel name which indicates that
	// channel is private.
	ChannelPrivatePrefix string
	// ChannelNamespaceBoundary is a string separator which must be put after
	// namespace part in channel name.
	ChannelNamespaceBoundary string
	// ChannelUserBoundary is a string separator which must be set before allowed
	// users part in channel name.
	ChannelUserBoundary string
	// ChannelUserSeparator separates allowed users in user part of channel name.
	ChannelUserSeparator string
	// ChannelMaxLength is a maximum length of channel name.
	ChannelMaxLength int
}

Config contains Application configuration options.

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates config and returns error if problems found

type ConnectEvent 

type ConnectEvent struct {
	Data Raw
}

ConnectEvent contains fields related to connect event.

type ConnectHandler 

type ConnectHandler func(context.Context, *Client, ConnectEvent) ConnectReply

ConnectHandler called when new client connects to server.

type ConnectReply 

type ConnectReply struct {
	Error      *Error
	Disconnect *Disconnect
	Data       Raw
}

ConnectReply contains fields determining the reaction on connect event.

type Credentials 

type Credentials struct {
	UserID   string
	ExpireAt int64
	Info     []byte
}

Credentials allows to authenticate connection when set into context.

type Disconnect 

type Disconnect struct {
	// Reason is a short description of disconnect.
	Reason string `json:"reason"`
	// Reconnect gives client an advice to reconnect after disconnect or not.
	Reconnect bool `json:"reconnect"`
}

Disconnect allows to configure how client will be disconnected from server.

type DisconnectEvent 

type DisconnectEvent struct {
	Disconnect *Disconnect
}

DisconnectEvent contains fields related to disconnect event.

type DisconnectHandler 

type DisconnectHandler func(DisconnectEvent) DisconnectReply

DisconnectHandler called when client disconnects from server.

type DisconnectReply 

type DisconnectReply struct{}

DisconnectReply contains fields determining the reaction on disconnect event.

type Encoding 

type Encoding = proto.Encoding

Encoding represents client connection transport encoding format.

type Engine 

type Engine interface {
	// contains filtered or unexported methods
}

Engine is responsible for PUB/SUB mechanics, channel history and presence information.

type EngineEventHandler 

type EngineEventHandler interface {
	// Publication must register callback func to handle Publications received.
	HandlePublication(ch string, pub *Publication) error
	// Join must register callback func to handle Join messages received.
	HandleJoin(ch string, join *Join) error
	// Leave must register callback func to handle Leave messages received.
	HandleLeave(ch string, leave *Leave) error
	// Control must register callback func to handle Control data received.
	HandleControl([]byte) error
}

EngineEventHandler can handle messages received from PUB/SUB system.

type Error 

type Error = proto.Error

Error represents client reply error.

type Hub 

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

Hub manages client connections.

func (*Hub) Channels 

func (h *Hub) Channels() []string

Channels returns a slice of all active channels.

func (*Hub) NumChannels 

func (h *Hub) NumChannels() int

NumChannels returns a total number of different channels.

func (*Hub) NumClients 

func (h *Hub) NumClients() int

NumClients returns total number of client connections.

func (*Hub) NumSubscribers 

func (h *Hub) NumSubscribers(ch string) int

NumSubscribers returns number of current subscribers for a given channel.

func (*Hub) NumUsers 

func (h *Hub) NumUsers() int

NumUsers returns a number of unique users connected.

type Info 

type Info struct {
	Nodes []NodeInfo
}

Info contains information about all known server nodes.

type Join 

type Join = proto.Join

Join sent to channel after someone subscribed.

type Leave 

type Leave = proto.Leave

Leave sent to channel after someone unsubscribed.

type LogEntry 

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

LogEntry represents log entry.

func NewLogEntry 

func NewLogEntry(level LogLevel, message string, fields ...map[string]interface{}) LogEntry

NewLogEntry creates new LogEntry.

type LogHandler 

type LogHandler func(LogEntry)

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

type LogLevel 

type LogLevel int

LogLevel describes the chosen log level. 

const (
	// LogLevelNone means no logging.
	LogLevelNone LogLevel = iota
	// LogLevelDebug turns on debug logs - its generally too much for production in normal
	// conditions but can help when developing and investigating problems in production.
	LogLevelDebug
	// LogLevelInfo is logs useful server information. This includes various information
	// about problems with client connections which is not Centrifugo errors but
	// in most situations malformed client behaviour.
	LogLevelInfo
	// LogLevelError level logs only server errors. This is logging that means non-working
	// Centrifugo and maybe effort from developers/administrators to make things
	// work again.
	LogLevelError
)

type MemoryEngine 

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

MemoryEngine allows to run Centrifugo without using Redis at all. All data managed inside process memory. With this engine you can only run single Centrifugo node. If you need to scale you should use Redis engine instead.

func NewMemoryEngine 

func NewMemoryEngine(n *Node, conf MemoryEngineConfig) (*MemoryEngine, error)

NewMemoryEngine initializes Memory Engine.

type MemoryEngineConfig 

type MemoryEngineConfig struct{}

MemoryEngineConfig is a memory engine config.

type MessageEvent 

type MessageEvent struct {
	Data Raw
}

MessageEvent contains fields related to message request.

type MessageHandler 

type MessageHandler func(MessageEvent) MessageReply

MessageHandler must handle incoming async message from client.

type MessageReply 

type MessageReply struct {
	Disconnect *Disconnect
}

MessageReply contains fields determining the reaction on message request.

type Node 

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

Node is a heart of centrifuge library – it internally keeps and manages client connections, maintains information about other centrifuge nodes, keeps useful references to things like engine, hub etc.

func New 

func New(c Config) (*Node, error)

New creates Node, the only required argument is config.

func (*Node) ChannelOpts 

func (n *Node) ChannelOpts(ch string) (ChannelOptions, bool)

ChannelOpts returns channel options for channel using current channel config.

func (*Node) Channels 

func (n *Node) Channels() ([]string, error)

Channels returns list of all channels currently active across on all nodes. This is a snapshot of state mostly useful for understanding what's going on with system.

func (*Node) Config 

func (n *Node) Config() Config

Config returns a copy of node Config.

func (*Node) Disconnect 

func (n *Node) Disconnect(user string, reconnect bool) error

Disconnect allows to close all user connections to Centrifugo.

func (*Node) History 

func (n *Node) History(ch string) ([]*Publication, error)

History returns a slice of last messages published into project channel.

func (*Node) Hub 

func (n *Node) Hub() *Hub

Hub returns node's Hub.

func (*Node) Info 

func (n *Node) Info() (Info, error)

Info returns aggregated stats from all nodes.

func (*Node) Log 

func (n *Node) Log(entry LogEntry)

Log allows to log entry.

func (*Node) NotifyShutdown 

func (n *Node) NotifyShutdown() chan struct{}

NotifyShutdown returns a channel which will be closed on node shutdown.

func (*Node) On 

func (n *Node) On() NodeEventHub

On allows access to NodeEventHub.

func (*Node) Presence 

func (n *Node) Presence(ch string) (map[string]*ClientInfo, error)

Presence returns a map with information about active clients in channel.

func (*Node) PresenceStats 

func (n *Node) PresenceStats(ch string) (PresenceStats, error)

PresenceStats returns presence stats from engine.

func (*Node) Publish 

func (n *Node) Publish(ch string, pub *Publication) error

Publish sends a message to all clients subscribed on channel. All running nodes will receive it and will send it to all clients on node subscribed on channel. If provided ChannelOptions is nil then Node will search for channel options automatically using configuration. If no channel options explicitly provided and no channel options found in configuration then this method will

func (*Node) PublishAsync 

func (n *Node) PublishAsync(ch string, pub *Publication) <-chan error

PublishAsync do the same as Publish but returns immediately after publishing message to engine. Caller can inspect error waiting for it on returned channel.

func (*Node) Reload 

func (n *Node) Reload(c Config) error

Reload node config.

func (*Node) RemoveHistory 

func (n *Node) RemoveHistory(ch string) error

RemoveHistory removes channel history.

func (*Node) Run 

func (n *Node) Run() error

Run performs node startup actions. At moment must be called once on start after engine set to Node.

func (*Node) SetEngine 

func (n *Node) SetEngine(e Engine)

SetEngine binds engine to node.

func (*Node) SetLogHandler 

func (n *Node) SetLogHandler(level LogLevel, handler LogHandler)

SetLogHandler sets LogHandler to handle log messages with severity higher than specific LogLevel.

func (*Node) Shutdown 

func (n *Node) Shutdown(ctx context.Context) error

Shutdown sets shutdown flag to Node so handlers could stop accepting new requests and disconnects clients with shutdown reason.

func (*Node) Unsubscribe 

func (n *Node) Unsubscribe(user string, ch string) error

Unsubscribe unsubscribes user from channel, if channel is equal to empty string then user will be unsubscribed from all channels.

type NodeEventHub 

type NodeEventHub interface {
	Connect(handler ConnectHandler)
}

NodeEventHub can deal with events binded to Node. All its methods are not goroutine-safe as handlers must be registered once before Node Run method called.

type NodeInfo 

type NodeInfo struct {
	UID         string
	Name        string
	Version     string
	NumClients  uint32
	NumUsers    uint32
	NumChannels uint32
	Uptime      uint32
}

NodeInfo contains information about node.

type PresenceStats 

type PresenceStats struct {
	NumClients int
	NumUsers   int
}

PresenceStats represents a short presence information for channel.

type Publication 

type Publication = proto.Publication

Publication allows to deliver custom payload to all channel subscribers.

type PublishEvent 

type PublishEvent struct {
	Channel string
	*Publication
}

PublishEvent contains fields related to publish event.

type PublishHandler 

type PublishHandler func(PublishEvent) PublishReply

PublishHandler called when client publishes into channel.

type PublishReply 

type PublishReply struct {
	Error      *Error
	Disconnect *Disconnect
}

PublishReply contains fields determining the reaction on publish event.

type RPCEvent 

type RPCEvent struct {
	Data Raw
}

RPCEvent contains fields related to rpc request.

type RPCHandler 

type RPCHandler func(RPCEvent) RPCReply

RPCHandler must handle incoming command from client.

type RPCReply 

type RPCReply struct {
	Error      *Error
	Disconnect *Disconnect
	Data       Raw
}

RPCReply contains fields determining the reaction on rpc request.

type Raw 

type Raw = proto.Raw

Raw represents raw bytes.

type RedisEngine 

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

RedisEngine uses Redis datastructures and PUB/SUB to manage Centrifugo logic. This engine allows to scale Centrifugo - you can run several Centrifugo instances connected to the same Redis and load balance clients between instances.

func NewRedisEngine 

func NewRedisEngine(n *Node, config RedisEngineConfig) (*RedisEngine, error)

NewRedisEngine initializes Redis Engine.

type RedisEngineConfig 

type RedisEngineConfig struct {
	Shards []RedisShardConfig
}

RedisEngineConfig of Redis Engine.

type RedisShardConfig 

type RedisShardConfig struct {
	// Host is Redis server host.
	Host string
	// Port is Redis server port.
	Port int
	// Password is password to use when connecting to Redis database. If empty then password not used.
	Password string
	// DB is Redis database number. If not set then database 0 used.
	DB int
	// Whether to use TLS connection or not.
	UseTLS bool
	// Whether to skip hostname verification as part of TLS handshake.
	TLSSkipVerify bool
	// Connection TLS configuration.
	TLSConfig *tls.Config
	// MasterName is a name of Redis instance master Sentinel monitors.
	MasterName string
	// SentinelAddrs is a slice of Sentinel addresses.
	SentinelAddrs []string
	// Prefix to use before every channel name and key in Redis.
	Prefix string
	// IdleTimeout is timeout after which idle connections to Redis will be closed.
	IdleTimeout time.Duration
	// PubSubNumWorkers sets how many PUB/SUB message processing workers will be started.
	// By default we start runtime.NumCPU() workers.
	PubSubNumWorkers int
	// ReadTimeout is a timeout on read operations. Note that at moment it should be greater
	// than node ping publish interval in order to prevent timing out Pubsub connection's
	// Receive call.
	ReadTimeout time.Duration
	// WriteTimeout is a timeout on write operations.
	WriteTimeout time.Duration
	// ConnectTimeout is a timeout on connect operation.
	ConnectTimeout time.Duration
}

RedisShardConfig is struct with Redis Engine options.

type RefreshEvent 

type RefreshEvent struct{}

RefreshEvent contains fields related to refresh event.

type RefreshHandler 

type RefreshHandler func(RefreshEvent) RefreshReply

RefreshHandler called when it's time to validate client connection and update it's expiration time.

type RefreshReply 

type RefreshReply struct {
	ExpireAt int64
	Info     Raw
}

RefreshReply contains fields determining the reaction on refresh event.

type SockjsConfig 

type SockjsConfig struct {
	// HandlerPrefix sets prefix for SockJS handler endpoint path.
	HandlerPrefix string

	// URL is URL address to SockJS client javascript library.
	URL string

	// HeartbeatDelay sets how often to send heartbeat frames to clients.
	HeartbeatDelay time.Duration

	// WebsocketReadBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	WebsocketReadBufferSize int

	// WebsocketWriteBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	WebsocketWriteBufferSize int
}

SockjsConfig represents config for SockJS handler.

type SockjsHandler 

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

SockjsHandler accepts SockJS connections.

func NewSockjsHandler 

func NewSockjsHandler(n *Node, c SockjsConfig) *SockjsHandler

NewSockjsHandler creates new SockjsHandler.

func (*SockjsHandler) ServeHTTP 

func (s *SockjsHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request)

type SubRefreshEvent 

type SubRefreshEvent struct {
	Channel string
}

SubRefreshEvent contains fields related to subscription refresh event.

type SubRefreshHandler 

type SubRefreshHandler func(SubRefreshEvent) SubRefreshReply

SubRefreshHandler called when it's time to validate client subscription to channel and update it's state if needed.

type SubRefreshReply 

type SubRefreshReply struct {
	Expired  bool
	ExpireAt int64
	Info     Raw
}

SubRefreshReply contains fields determining the reaction on subscription refresh event.

type SubscribeEvent 

type SubscribeEvent struct {
	Channel string
}

SubscribeEvent contains fields related to subscribe event.

type SubscribeHandler 

type SubscribeHandler func(SubscribeEvent) SubscribeReply

SubscribeHandler called when client wants to subscribe on channel.

type SubscribeReply 

type SubscribeReply struct {
	Error       *Error
	Disconnect  *Disconnect
	ExpireAt    int64
	ChannelInfo Raw
}

SubscribeReply contains fields determining the reaction on subscribe event.

type Transport 

type Transport interface {
	// Name returns a name of transport used for client connection.
	Name() string
	// Encoding returns transport encoding used.
	Encoding() Encoding
	// Info returns transport information.
	Info() TransportInfo
}

Transport abstracts a connection transport between server and client.

type TransportInfo 

type TransportInfo struct {
	// Request contains initial HTTP request sent by client. Can be nil in case of
	// non-HTTP based transports. Though both Websocket and SockjS we currently
	// support use HTTP on start so this field will present.
	Request *http.Request
}

TransportInfo contains extended transport description.

type UnsubscribeEvent 

type UnsubscribeEvent struct {
	Channel string
}

UnsubscribeEvent contains fields related to unsubscribe event.

type UnsubscribeHandler 

type UnsubscribeHandler func(UnsubscribeEvent) UnsubscribeReply

UnsubscribeHandler called when client unsubscribed from channel.

type UnsubscribeReply 

type UnsubscribeReply struct {
}

UnsubscribeReply contains fields determining the reaction on unsubscribe event.

type WebsocketConfig 

type WebsocketConfig struct {
	// Compression allows to enable websocket permessage-deflate
	// compression support for raw websocket connections. It does
	// not guarantee that compression will be used - i.e. it only
	// says that server will try to negotiate it with client.
	Compression bool

	// CompressionLevel sets a level for websocket compression.
	// See posiible value description at https://golang.org/pkg/compress/flate/#NewWriter
	CompressionLevel int

	// CompressionMinSize allows to set minimal limit in bytes for
	// message to use compression when writing it into client connection.
	// By default it's 0 - i.e. all messages will be compressed when
	// WebsocketCompression enabled and compression negotiated with client.
	CompressionMinSize int

	// ReadBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	ReadBufferSize int

	// WriteBufferSize is a parameter that is used for raw websocket Upgrader.
	// If set to zero reasonable default value will be used.
	WriteBufferSize int

	// CheckOrigin func to provide custom origin check logic.
	// nil means allow all origins.
	CheckOrigin func(r *http.Request) bool
}

WebsocketConfig represents config for WebsocketHandler.

type WebsocketHandler 

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

WebsocketHandler handles websocket client connections.

func NewWebsocketHandler 

func NewWebsocketHandler(n *Node, c WebsocketConfig) *WebsocketHandler

NewWebsocketHandler creates new WebsocketHandler.

func (*WebsocketHandler) ServeHTTP 

func (s *WebsocketHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request)

Source Files

View all Source files

Directories

Path Synopsis
examples
benchmark
chat_json
chat_oauth2
chat_protobuf
redis_engine
testsuite
worms
internal
auth
Package auth provides functions to generate and check Centrifugo auth signs.
 Package auth provides functions to generate and check Centrifugo auth signs.
priority
Package priority provides priority queue for libcentrifugo package Memory Engine.
 Package priority provides priority queue for libcentrifugo package Memory Engine.
proto
Package proto is a generated protocol buffer package.
 Package proto is a generated protocol buffer package.
proto/controlproto
Package controlproto is a generated protocol buffer package.
 Package controlproto is a generated protocol buffer package.
queue
timers
uuid
Package uuid provides implementation of Universally Unique Identifier (UUID).
 Package uuid provides implementation of Universally Unique Identifier (UUID).

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.