wsq

Documentation

Overview

Package wsq is a lightweight framework for building a websocket server that communicates with the NSQ messaging platform.

Simple Server

Creating the default server

server := wsq.NewDefaultServer(":9980")
server.Run()

This creates a websocket server listening on port "9980". The default server has no authentication enabled and no marshaling configured. It connects to a default NSQ instance address and relays messages as is. See wsq.Config docs for all default values being used.

Full-featured Server

The default server isn't that useful in production.

Here is an example of `wsq.NewServer` usage

// Create the default config
wsqConfig := wsq.NewConfig()
// Configure Websocket CheckOrigin callback to bypass any checks.
// Don't do it in production!
wsqConfig.SetWSCheckOrigin(wsq.CheckOriginBypass)

// Create the server instance specifying message and user types to use.
server := wsq.NewServer[message, *wsq.AnonymousUser](
	// Address (optional) and port to listen on
	":9980",
	// WSQ Config instance
	wsqConfig,
	// NSQ Config instance
	nsq.NewConfig(),
	// WSQ Transformer struct providing encoders/decoders for NSQ and Websocket sides respectivly
	&wsq.Transformer[message]{NSQEnDec: &nsqEnDec{}, WSEnDec: &wsEnDec{}},
	// Authentication controller
	&wsq.NoAuthentication,
)
server.Run()

See wsq.NewServer for detailed documentation on creating WSQ Server.

See full server example in the demo directory of the WSQ repo: https://codeberg.org/lig/wsq/src/branch/main/_demo/srv/cmd/demo/main.go

Protocol

WSQ uses a protocol somewhat similar to NSQ protocol to communicate with a client connected over the websocket.

Each websocket message from the client to WSQ has the following form:

<command> <args>

where

• <command> -- one of `SUB`, `USUB`, `PUB`;
• <args> -- argument list depending on the command.

Commands:

SUB <topicName>

	<topicName> -- The Name of the NSQ topic to subscribe.

USUB <topicName>

	<topicName> -- The Name of the NSQ topic to unsubscribe.

PUB <topicName> <message>

	<topicName> -- The Name of the NSQ topic to publish message to;
	<message> -- The message to publish.

The <message> provided in `PUB` command is subject to marshaling, see “Message Encoding and Decoding” below.

There are two types of websocket messages from WSQ to the client:

• normal message;
• error message.

The normal websocket message from WSQ to the client has the following form:

<topicName> <message>

where

• <topicName> -- the name of the NSQ topic of the received message;
• <message> -- the message received from NSQ.

The <message> here is subject to marshaling, see “Message Encoding and Decoding” below.

The error websocket message from WSQ to the client has the following form:

#err <code> <description>

where

• <code> -- the 3-digit code representing the error type similar to HTTP response codes;
• <description> -- human readable error description useful for debugging.

As of the time of writing this documentation there are 3 error codes defined in WSQ:

• 400 -- errors related to malformed client commands;
• 403 -- errors related to unauthorized access to some actions;
• 500 -- errors related to communicating to NSQ or other server-related errors.

Message Encoding and Decoding

WSQ provides an ability to marshal and unmarshal messages on NSQ and Websocket sides. For each side a type implementing the wsq.EnDec interface must be defined. A message received from NSQ will be unmarshalled using `[EnDec].UnMarshal` defined for the NSQ side and marshaled using `[EnDec].Marshal` defined for the Websocket side before being send to a client connected over the websocket. The same works in the opposite direction with corresponding EnDecs being used for unmarshalling and marshalling.

Authentication and authorization

WSQ provides wsq.Authenticator and wsq.User interfaces that must be implemented for authentication and authorization to work respectively. It is up to developer how to implement those interfaces.

The `[Authenticator].Authenticate` method is being called before WSQ starts any client command processing. The method must return a valid User instance for WSQ to accept the client. There is no requirement for `[User].IsAuthenticated` method to return `true` if called.

For convenience WSQ provides the AuthMessageAuthenticator type and the AuthMessageHandler interface that allow to implement authentication using a special `AUTH` client command.

The User interface is being used for authorizing access to subscribing and publishing for a specific topic.

Message filtering

WSQ provides an ability to filter incoming messages basing on the NSQ topic name and message content. The User interface provides the `CanRead` method for this purpose.

Index

• Variables
• func CheckOriginBypass(r *http.Request) bool
• type AnonymousAuthenticator
  • func (a *AnonymousAuthenticator) Authenticate(wsConn wsConnI, r *http.Request) (*AnonymousUser, error)
• type AnonymousUser
  • func NewAnonymousUser(grantAll bool) *AnonymousUser
  • func (u *AnonymousUser) CanPub(topic string) bool
  • func (u *AnonymousUser) CanRead(topic string, message any) bool
  • func (u *AnonymousUser) CanSub(topic string) bool
  • func (u *AnonymousUser) IsAuthenticated() bool
• type AuthMessageAuthenticator
  • func NewAuthMessageAuthenticator[U User](handler AuthMessageHandler[U]) *AuthMessageAuthenticator[U]
  • func (a *AuthMessageAuthenticator[U]) Authenticate(wsConn wsConnI, r *http.Request) (User, error)
• type AuthMessageHandler
• type AuthenticationError
  • func (e *AuthenticationError) Error() string
• type Authenticator
• type BypassEnDec
  • func (b BypassEnDec) Marshal(topic string, msg []byte) ([]byte, error)
  • func (b BypassEnDec) UnMarshal(topic string, data []byte) ([]byte, error)
• type Config
  • func NewConfig() *Config
  • func (c *Config) SetNSQChannelPrefix(value string)
  • func (c *Config) SetNSQDAddr(value string)
  • func (c *Config) SetNSQLookupdAddrs(value []string)
  • func (c *Config) SetWSCheckOrigin(value func(r *http.Request) bool)
  • func (c *Config) SetWSWriteTimeout(value time.Duration)
• type EnDec
• type Server
  • func NewDefaultServer(addr string) *Server[[]byte, *AnonymousUser]
  • func NewServer[M any, U User](addr string, wsqConfig *Config, nsqConfig *nsq.Config, ...) *Server[M, U]
  • func (s *Server[M, U]) Run()
• type Transformer
• type User

Variables

var DummyTransformer = Transformer[[]byte]{
	NSQEnDec: &BypassEnDec{},
	WSEnDec:  &BypassEnDec{},
}

A Transformer instance that uses BypassEnDec on both sides.

var NoAuthentication = AnonymousAuthenticator{}

NoAuthentication is a default AnonymousAuthenticator instance provided for convenience.

Functions

func CheckOriginBypass

func CheckOriginBypass(r *http.Request) bool

Types

type AnonymousAuthenticator

type AnonymousAuthenticator struct{}

AnonymousAuthenticator is an Authenticator implementation that allows full anonymous access to a WSQ server without authentication.

func (*AnonymousAuthenticator) Authenticate

func (a *AnonymousAuthenticator) Authenticate(wsConn wsConnI, r *http.Request) (*AnonymousUser, error)

type AnonymousUser

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

AnonymousUser type implements User and can be used to allow full access or deny any access, see NewAnonymousUser.

func NewAnonymousUser

func NewAnonymousUser(grantAll bool) *AnonymousUser

NewAnonymousUser creates new AnonymousUser instance. Use `grantAll` argument to control new instance behavior. If `grantAll` is `true` then all methods defined in User return `true` and wise versa.

func (*AnonymousUser) CanPub

func (u *AnonymousUser) CanPub(topic string) bool

func (*AnonymousUser) CanRead

func (u *AnonymousUser) CanRead(topic string, message any) bool

func (*AnonymousUser) CanSub

func (u *AnonymousUser) CanSub(topic string) bool

func (*AnonymousUser) IsAuthenticated

func (u *AnonymousUser) IsAuthenticated() bool

type AuthMessageAuthenticator

type AuthMessageAuthenticator[U User] struct {
	AuthMessageHandler[U]
}

AuthMessageAuthenticator is a generic implementation of the Authenticator interface that uses AuthMessageHandler implementation to authenticate websocket client using a special `AUTH` command.

The `AUTH` websocket command has the form `AUTH <auth_data>`. AuthMessageAuthenticator handles processing of this command, passes `<auth_data>` as is to the provided [AuthMessageHandler.AuthenticateViaAuthMessage] implementation, and returns its result as is.

func NewAuthMessageAuthenticator

func NewAuthMessageAuthenticator[U User](handler AuthMessageHandler[U]) *AuthMessageAuthenticator[U]

func (*AuthMessageAuthenticator[U]) Authenticate

func (a *AuthMessageAuthenticator[U]) Authenticate(wsConn wsConnI, r *http.Request) (User, error)

type AuthMessageHandler

type AuthMessageHandler[U User] interface {
	AuthenticateViaAuthMessage(message []byte) (U, error)
}

AuthMessageHandler is an interface defining AuthenticateViaAuthMessage method used in AuthMessageAuthenticator for handling authentication message data.

This is a generic interface. The implementation must be instantiated using type `U` which must implement User.

type AuthenticationError

type AuthenticationError struct {
	Message string
}

func (*AuthenticationError) Error

func (e *AuthenticationError) Error() string

type Authenticator

type Authenticator[U User] interface {
	// Authenticate receives [websocket.Conn] and [http.Request], can communicate on
	// websocket.
	// Must return [User] on successful authentication.
	// Must return [AuthenticationError] in case of authentication error.
	// Should not close websocket connection.
	Authenticate(wsConn wsConnI, r *http.Request) (U, error)
}

Authenticator defines interface for implementing authentication in WSQ.

This is a generic interface. The implementation must be instantiated using type `U` that must implement the User interface.

type BypassEnDec

type BypassEnDec struct{}

An EnDec implementation that uses `[]byte` type as message representation and doesn't perform any transformation on the data itself.

func (BypassEnDec) Marshal

func (b BypassEnDec) Marshal(topic string, msg []byte) ([]byte, error)

func (BypassEnDec) UnMarshal

func (b BypassEnDec) UnMarshal(topic string, data []byte) ([]byte, error)

type Config

type Config struct {
	// Producer NSQD Address in the form "<host>:<port>". The default is "nsqd:4150".
	NSQDAddr string
	// A list of NSQ Lookupd addresses ("<host>:<port>") to be used for Consumer.
	// Uses `NSQDAddr` if `NSQLookupdAddrs` is empty (the default)
	NSQLookupdAddrs []string
	// The prefix to use for channel names being created by WSQ. The default is "WSQ".
	NSQChannelPrefix string
	// See gorilla/websocket docs for Upgrader: https://pkg.go.dev/github.com/gorilla/websocket@v1.5.0?utm_source=gopls#Upgrader.CheckOrigin
	// The default is `nil`.
	WSCheckOrigin func(r *http.Request) bool
	// Websocket write timeout. Zero duration means no timeout. The default is 10 seconds.
	WSWriteTimeout time.Duration
}

Config struct holds wsq-specific configuration values. Use NewConfig to build a new instance and alter the configuration using setter methods as needed.

See nsq.Config docs for nsq-specific configuration.

func NewConfig

func NewConfig() *Config

NewConfig function creates a new wsq `Config` instance and returns its pointer.

func (*Config) SetNSQChannelPrefix

func (c *Config) SetNSQChannelPrefix(value string)

func (*Config) SetNSQDAddr

func (c *Config) SetNSQDAddr(value string)

func (*Config) SetNSQLookupdAddrs

func (c *Config) SetNSQLookupdAddrs(value []string)

func (*Config) SetWSCheckOrigin

func (c *Config) SetWSCheckOrigin(value func(r *http.Request) bool)

func (*Config) SetWSWriteTimeout

func (c *Config) SetWSWriteTimeout(value time.Duration)

type EnDec

type EnDec[M any] interface {
	Marshal(topic string, msg M) ([]byte, error)
	UnMarshal(topic string, data []byte) (M, error)
}

Defines interface for handling serialization separately on each of NSQ and Websocket sides.

This is a generic interface. The implementation must be instantiated using type `M` which coordinates Marshal method input and UnMarshal method output types.

type Server

type Server[M any, U User] struct {
	// contains filtered or unexported fields
}

func NewDefaultServer

func NewDefaultServer(addr string) *Server[[]byte, *AnonymousUser]

Creates new WSQ server using defaults for Config, nsq.Config, Transformer, and Authenticator.

func NewServer

func NewServer[M any, U User](
	addr string,
	wsqConfig *Config,
	nsqConfig *nsq.Config,
	transformer *Transformer[M],
	authenticator Authenticator[U],
) *Server[M, U]

Creates new WSQ server.

This function is generic and must be instantiated using types `M` representing the type for message data in between unmarshalling and marshalling (see Transformer) and `U` representing User interface implementation (see Authenticator).

func (*Server[M, U]) Run

func (s *Server[M, U]) Run()

type Transformer

type Transformer[M any] struct {
	NSQEnDec EnDec[M]
	WSEnDec  EnDec[M]
}

Stores NSQ and Websocket EnDec interface implementations.

This is a generic struct. Its instance must be instantiated using type `M` which coordinates Marshal method input and UnMarshal method output types across both EnDecs.

type User

type User interface {
	// IsAuthenticated returns user's authentication status. This method is provided for
	// convenience only. Other User methods could utilize IsAuthenticated result. There
	// is no internal requirement for it to return `true` for user to be granted
	// access to communicating with NSQ.
	IsAuthenticated() bool
	// CanPub grants user access to publishing to the specified topic.
	CanPub(topic string) bool
	// CanSub grants user access to subscribing to the specified topic.
	CanSub(topic string) bool
	// CanRead grants user access to receiving the specified message posted to the
	// specified topic.
	CanRead(topic string, message any) bool
}