pubsub

README

pubsub

golang websocket pubsub server and client

Built using the gorilla/websocket library.

Documentation

Index

• Constants
• Variables
• func Pub(ctx context.Context, host string, isSecure bool, path string, id string, ...) (err error)
• func PubWithBearerToken(ctx context.Context, host string, token string, isSecure bool, path string, ...) (err error)
• func PubWithHeader(parentCtx context.Context, host string, header http.Header, isSecure bool, ...) (err error)
• type Client
  • func NewClient(host string, isSecure bool, path string, id string) *Client
  • func NewClientWithBearerToken(host string, token string, isSecure bool, path string, id string) *Client
  • func NewClientWithHeader(host string, header http.Header, isSecure bool, path string, id string) *Client
  • func (c *Client) AllowInsecureConnections()
  • func (c *Client) ReadMessage() (messageType int, p []byte, err error)
  • func (c *Client) Start(parentCtx context.Context, messageFunc func(message []byte) (err error)) (err error)
  • func (c *Client) WriteJSON(v interface{}) (err error)
  • func (c *Client) WriteMessage(messageType int, data []byte) (err error)
  • func (c *Client) WriteMessageSlice(dataSlice [][]byte) (err error)
  • func (c *Client) WriteTextMessage(data []byte) (err error)
• type Server
  • func NewServer(ctx context.Context) *Server
  • func NewServerWithSettings(ctx context.Context, writeWait time.Duration, pongWait time.Duration, ...) *Server
  • func (s *Server) HandlePub(channel string, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandlePubSub(channel string, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandlePubSubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandlePubSubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, ...) (err error)
  • func (s *Server) HandlePubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandlePubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, ...) (err error)
  • func (s *Server) HandleSub(channel string, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandleSubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)
  • func (s *Server) HandleSubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, ...) (err error)
• type Store

Constants

const (
	BYTE     = 1 << (iota * 10) // = 1 << 0
	KILOBYTE                    // = 1 << 10
	MEGABYTE                    // = 1 << 20
	GIGABYTE                    // = 1 << 30

	// Gorilla Message types
	TextMessage   = websocket.TextMessage
	BinaryMessage = websocket.BinaryMessage
	CloseMessage  = websocket.CloseMessage
	PingMessage   = websocket.PingMessage
	PongMessage   = websocket.PongMessage
)

Variables

var BroadcastLimitExceededCloseMessage = websocket.FormatCloseMessage(websocket.ClosePolicyViolation, "client broadcast message limit exceeded")

var ErrStoreLimitExceeded = errors.New("websockets store limit exceeded")

Functions

func Pub

func Pub(
	ctx context.Context,
	host string,
	isSecure bool,
	path string,
	id string,
	message []byte,
) (err error)

Pub creates a new websockets client Pub publishes a message and then closes the client connection

func PubWithBearerToken

func PubWithBearerToken(
	ctx context.Context,
	host string,
	token string,
	isSecure bool,
	path string,
	id string,
	message []byte,
) (err error)

PubWithBearerToken creates a new websockets client with a brearer token PubWithBearerToken publishes a message and then closes the client connection

func PubWithHeader

func PubWithHeader(
	parentCtx context.Context,
	host string,
	header http.Header,
	isSecure bool,
	path string,
	id string,
	message []byte,
) (err error)

PubWithHeader creates a new websockets client with a custom header PubWithHeader publishes a message and then closes the client connection

Types

type Client

type Client struct {

	// Settings
	WriteWait        time.Duration // Time allowed to write a message to the peer.
	PongWait         time.Duration // Time allowed to read the next pong message from the peer.
	PingPeriod       time.Duration // Send pings to peer with this period. Must be less than PongWait.
	MaxMessageSize   int64         // Maximum message size allowed from peer.
	CloseGracePeriod time.Duration // Grace period after closing websocket.

	Dialer *websocket.Dialer
	// contains filtered or unexported fields
}

Client is the websockets client

func NewClient

func NewClient(
	host string,
	isSecure bool,
	path string,
	id string,
) *Client

NewClient creates a new websockets client

func NewClientWithBearerToken

func NewClientWithBearerToken(
	host string,
	token string,
	isSecure bool,
	path string,
	id string,
) *Client

NewClientWithBearerToken creates a new websockets client with a brearer token

func NewClientWithHeader

func NewClientWithHeader(
	host string,
	header http.Header,
	isSecure bool,
	path string,
	id string,
) *Client

NewClientWithHeader creates a new websockets client with a custom header

func (*Client) AllowInsecureConnections

func (c *Client) AllowInsecureConnections()

AllowInsecureConnections allows the client to make insecure connections (i.e. to servers with self-signed certificates)

func (*Client) ReadMessage

func (c *Client) ReadMessage() (messageType int, p []byte, err error)

ReadMessage reads messageType int and p []byte from the websocket connection

func (*Client) Start

func (c *Client) Start(parentCtx context.Context, messageFunc func(message []byte) (err error)) (err error)

Start starts the websocket client

func (*Client) WriteJSON

func (c *Client) WriteJSON(v interface{}) (err error)

WriteJSON writes marshalls v interface{} as JSON and sends the json []byte to the websocket connection

func (*Client) WriteMessage

func (c *Client) WriteMessage(messageType int, data []byte) (err error)

WriteMessage writes b []byte to the websocket connection

func (*Client) WriteMessageSlice

func (c *Client) WriteMessageSlice(dataSlice [][]byte) (err error)

WriteMessageSlice writes each set of bytes in the dataSlice [][]byte to one websocket connection

func (*Client) WriteTextMessage

func (c *Client) WriteTextMessage(data []byte) (err error)

WriteTextMessage writesa websocket.TextMessage b []byte message to the websocket connection

type Server

type Server struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Server is the websockets server struct containing m (a map[string] of map[*Store]bool entries)

func NewServer

func NewServer(ctx context.Context) *Server

NewServer returns a new websockets server with default settings

func NewServerWithSettings

func NewServerWithSettings(
	ctx context.Context,
	writeWait time.Duration,
	pongWait time.Duration,
	pingPeriod time.Duration,
	maxMessageSize int64,
) *Server

NewServerWithSettings returns a new websockets server

func (*Server) HandlePub

func (s *Server) HandlePub(channel string, w http.ResponseWriter, r *http.Request) (err error)

HandlePub is HandlePub but is Pub only (i.e. broadcast messages sent by other clients are not received by this websocket connection)

func (*Server) HandlePubSub

func (s *Server) HandlePubSub(channel string, w http.ResponseWriter, r *http.Request) (err error)

HandlePubSub is a handler that creates a websocket connection with a channel string identifier All messages sent via this websocket connection are broadcast to all other clients subscribed with this channel identifier All broadcast messages sent by other clients using this identifier are received by this websocket connection

func (*Server) HandlePubSubWithHeader

func (s *Server) HandlePubSubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)

HandlePubSubWithHeader is HandlePubSub but uses a custom header

func (*Server) HandlePubSubWithHeaderAndLimits

func (s *Server) HandlePubSubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, broadcastLimit time.Duration, w http.ResponseWriter, r *http.Request) (err error)

HandlePubSubWithHeaderAndLimits is HandlePubSub but uses a custom header and has store and/or broadcast limits storeLimit is the maximum number of concurrent Stores (i.e. connections) that is allowed for each channel (default is 0 - no limit) broadcastLimit is the minimum delay required between broadcasts for each connection (default is 0*time.Second - no limit) The connection is closed if the broadcast limit is exceeded

func (*Server) HandlePubWithHeader

func (s *Server) HandlePubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)

HandlePubWithHeader is HandlePub but uses a custom header

func (*Server) HandlePubWithHeaderAndLimits

func (s *Server) HandlePubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, broadcastLimit time.Duration, w http.ResponseWriter, r *http.Request) (err error)

HandlePubWithHeaderAndLimits is HandlePub but uses a custom header and has store and/or broadcast limits storeLimit is the maximum number of concurrent Stores (i.e. connections) that is allowed for each channel (default is 0 - no limit) broadcastLimit is the minimum delay required between broadcasts for each connection (default is 0*time.Second - no limit) The connection is closed if the broadcast limit is exceeded

func (*Server) HandleSub

func (s *Server) HandleSub(channel string, w http.ResponseWriter, r *http.Request) (err error)

HandleSub is HandlePubSub but is Sub only (i.e. messages sent via this websocket connection are not broadcast to other subscribed clients)

func (*Server) HandleSubWithHeader

func (s *Server) HandleSubWithHeader(channel string, header http.Header, w http.ResponseWriter, r *http.Request) (err error)

HandleSubWithHeader is HandleSub but uses a custom header

func (*Server) HandleSubWithHeaderAndLimits

func (s *Server) HandleSubWithHeaderAndLimits(channel string, header http.Header, storeLimit int64, broadcastLimit time.Duration, w http.ResponseWriter, r *http.Request) (err error)

HandleSubWithHeaderAndLimits is HandleSub but uses a custom header and has store and/or broadcast limits storeLimit is the maximum number of concurrent Stores (i.e. connections) that is allowed for each channel (default is 0 - no limit) broadcastLimit is the minimum delay required between broadcasts for each connection (default is 0*time.Second - no limit) The connection is closed if the broadcast limit is exceeded

type Store

type Store struct {
	sync.Mutex // write lock
	// contains filtered or unexported fields
}

Store stores a conn variable (the *websocket.Conn) and a sendCh variable (a buffered channel of outbound []byte messages)