protocol

Documentation

Index

• Constants
• func DefaultEncoder(ctx context.Context, data interface{}, err error, pw *PushWriter) error
• func ErrRPCNotFound(rpcName string) error
• type Config
  • func NewConfig() Config
  • func (c Config) Validate() error
• type Connection
  • func NewConnection(ctx context.Context, conn *websocket.Conn) (*Connection, error)
• type Connections
  • func NewConnections(s *statsd.Client, l log.Logger) *Connections
  • func (c *Connections) Authenticate(writer *PushWriter, session *model.Session)
  • func (c *Connections) Count() int
  • func (c *Connections) Deauthenticate(writer *PushWriter)
  • func (c *Connections) Deregister(writer *PushWriter)
  • func (c *Connections) EncodeTo(userID globalid.ID, m *Message) error
  • func (c *Connections) Register(ctx context.Context, conn *websocket.Conn) (*PushWriter, error)
  • func (c *Connections) Writers() []*PushWriter
  • func (c *Connections) WritersByUser(userID globalid.ID) []*PushWriter
• type Error
  • func (e *Error) Error() string
  • func (e Error) MarshalJSON() ([]byte, error)
  • func (e *Error) UnmarshalJSON(b []byte) error
• type Message
  • func NewMessage(rpc string) *Message
  • func (m *Message) Encode(ctx context.Context, w io.Writer) error
  • func (m *Message) EncodePayload(data interface{}) error
  • func (m *Message) Error() string
• type MessageEndpoint
• type Middleware
• type PushWriter
  • func NewPushWriter(conn *Connection, w io.Writer, log log.Logger) *PushWriter
  • func (pw *PushWriter) Authenticated() bool
  • func (pw *PushWriter) Connection() *Connection
  • func (pw *PushWriter) Session() *model.Session
  • func (pw *PushWriter) SetSession(s *model.Session)
  • func (pw *PushWriter) Write(data []byte) (int, error)
• type Router
  • func NewRouter(s *store.Store, conns *Connections, settings *model.Settings, c *Config, ...) *Router
  • func (r *Router) Close(c io.Closer)
  • func (r *Router) Connections() *Connections
  • func (r *Router) HandleConnection(ctx context.Context, conn *websocket.Conn, req *http.Request) error
  • func (r *Router) RegisterRPC(rpcName string, e MessageEndpoint)
  • func (r *Router) Route(ctx context.Context, session *model.Session, writer *PushWriter, ...) error
  • func (r *Router) UpgradeConnection() http.HandlerFunc
• type WebSocketWriter
  • func (wsw WebSocketWriter) Write(data []byte) (int, error)

Constants

const (
	// RequestID is a unique identifier for every request in order to
	// implement distributed request tracing.
	RequestID = "RequstID"
	// SessionID is used in order to retrieve the current session for the given
	// context.
	SessionID = "SessionID"
	// Callback is used to identify and match up the RPC of the original
	// request. It is helpful for logging and the client.
	Callback = "Callback"
	// IPAddress is the context key used to identify the IP address of a
	// connection.
	IPAddress = "IPAddress"
	// IPAddress is the context key used to identify the user agent of a
	// connection.
	UserAgent = "UserAgent"
)

Functions

func DefaultEncoder

func DefaultEncoder(ctx context.Context, data interface{}, err error, pw *PushWriter) error

DefaultEncoder encodes the result and writes it to the given push writer.

DefaultEncoder provides a default way of handling the response writing. The Request ID is used as a value in the ack field of the message. The original RPC, if this message is a response to a request, is used as a value in the callback field.

func ErrRPCNotFound

func ErrRPCNotFound(rpcName string) error

ErrRPCNotFound is used when a request with an unknown or non-registered RPC was made.

Types

type Config

type Config struct {
	EnableBugSnag bool
}

func NewConfig

func NewConfig() Config

func (Config) Validate

func (c Config) Validate() error

Validate returns an error if the config is invalid. TODO: (corylanou) add validation

type Connection

type Connection struct {
	IPAddress  string         `json:"ipAddress"`
	UserAgent  string         `json:"userAgent"`
	AdminPanel bool           `json:"adminPanel"`
	Session    *model.Session `json:"session"`
	CreatedAt  time.Time      `json:"createdAt"`
	// contains filtered or unexported fields
}

func NewConnection

func NewConnection(ctx context.Context, conn *websocket.Conn) (*Connection, error)

type Connections

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

Connections manages PushWriter by making them accessible by their user ID if authenticated.

func NewConnections

func NewConnections(s *statsd.Client, l log.Logger) *Connections

func (*Connections) Authenticate

func (c *Connections) Authenticate(writer *PushWriter, session *model.Session)

func (*Connections) Count

func (c *Connections) Count() int

func (*Connections) Deauthenticate

func (c *Connections) Deauthenticate(writer *PushWriter)

func (*Connections) Deregister

func (c *Connections) Deregister(writer *PushWriter)

func (*Connections) EncodeTo

func (c *Connections) EncodeTo(userID globalid.ID, m *Message) error

func (*Connections) Register

func (c *Connections) Register(ctx context.Context, conn *websocket.Conn) (*PushWriter, error)

func (*Connections) Writers

func (c *Connections) Writers() []*PushWriter

func (*Connections) WritersByUser

func (c *Connections) WritersByUser(userID globalid.ID) []*PushWriter

type Error

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

Error is a wrapper error in order to hide error information to the client.

func (*Error) Error

func (e *Error) Error() string

Error implements the error interface for Error.

func (Error) MarshalJSON

func (e Error) MarshalJSON() ([]byte, error)

MarshalJSON ensures errors are encoded in the form of strings.

func (*Error) UnmarshalJSON

func (e *Error) UnmarshalJSON(b []byte) error

UnmarshalJSON ensures errors are read as strings and unmarshaled in the error again.

type Message

type Message struct {
	RPC       string          `json:"rpc,omitempty"`
	SessionID globalid.ID     `json:"sessionID,omitempty"`
	RequestID globalid.ID     `json:"requestID,omitempty"`
	Ack       globalid.ID     `json:"ack,omitempty"`
	Data      json.RawMessage `json:"data,omitempty"`
	Err       string          `json:"error,omitempty"`
	Callback  string          `json:"callback,omitempty"`
}

Message is the basic building of the protocol. A message can be an action, a request and a corresponding response.

func NewMessage

func NewMessage(rpc string) *Message

func (*Message) Encode

func (m *Message) Encode(ctx context.Context, w io.Writer) error

func (*Message) EncodePayload

func (m *Message) EncodePayload(data interface{}) error

EncodePayload is a method for the client package to marshal the data part of the message.

func (*Message) Error

func (m *Message) Error() string

Error implements the error interface for Message.

type MessageEndpoint

type MessageEndpoint func(ctx context.Context, s *model.Session, pw *PushWriter, m Message) error

MessageEndpoint is the fundamental building block for the protocol on the receiving respectively serving end. It represents a single RPC method.

type Middleware

type Middleware func(endpoint MessageEndpoint) MessageEndpoint

Middleware is a building block which takes an endpoint and returns an endpoint again.

type PushWriter

type PushWriter struct {
	ID globalid.ID
	// contains filtered or unexported fields
}

PushWriter is a writer which is unique for every session, respectively for every unauthenticated request. Besides session it tracks callbacks which have been registered to the message bus.

func NewPushWriter

func NewPushWriter(conn *Connection, w io.Writer, log log.Logger) *PushWriter

NewPushWriter returns a new instance of PushWriter. An optional writer can be passed for testing purposes.

func (*PushWriter) Authenticated

func (pw *PushWriter) Authenticated() bool

func (*PushWriter) Connection

func (pw *PushWriter) Connection() *Connection

func (*PushWriter) Session

func (pw *PushWriter) Session() *model.Session

func (*PushWriter) SetSession

func (pw *PushWriter) SetSession(s *model.Session)

SetSession allows the session to be set via go routines

func (*PushWriter) Write

func (pw *PushWriter) Write(data []byte) (int, error)

Write is a write operation which can be used concurrently.

type Router

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

Router registers RPCs to be matched and dispatches a handler.

func NewRouter

func NewRouter(s *store.Store, conns *Connections, settings *model.Settings, c *Config, l log.Logger) *Router

NewRouter returns a new Router instance.

func (*Router) Close

func (r *Router) Close(c io.Closer)

Close helper to ensure the returned error is checked.

func (*Router) Connections

func (r *Router) Connections() *Connections

func (*Router) HandleConnection

func (r *Router) HandleConnection(ctx context.Context, conn *websocket.Conn, req *http.Request) error

HandleConnection handles incoming messages in accordance to the RPC protocol. Each RPC request is handled in its own go routine.

func (*Router) RegisterRPC

func (r *Router) RegisterRPC(rpcName string, e MessageEndpoint)

RegisterRPC adds a new endpoint to the list of available routes. It ensures some resilience to using clients by lower casing the RPC name.

func (*Router) Route

func (r *Router) Route(ctx context.Context, session *model.Session, writer *PushWriter, request string) error

Route is responsible for parsing incoming messages and invokes the endpoint after determening it based on the message header.

func (*Router) UpgradeConnection

func (r *Router) UpgradeConnection() http.HandlerFunc

UpgradeConnection upgrades the HTTP server connection to the WebSocket protocol and handles incoming messages in accordance to the RPC protocol.

type WebSocketWriter

type WebSocketWriter struct {
	Conn *websocket.Conn
}

WebSocketWriter is a writer wrapper for a given WebSocket connection.

func (WebSocketWriter) Write

func (wsw WebSocketWriter) Write(data []byte) (int, error)

Write is the basic operation for sending messages on the WebSocket connection.