wsutil

Documentation

Overview

Package wsutil provides abstractions around the Websocket, including rate limits.

Index

• Variables
• func ErrBrokenConnection(err error) error
• func HandleEvent(h EventHandler, ev Event) error
• func IsBrokenConnection(err error) bool
• func NewDialLimiter() *rate.Limiter
• func NewGlobalIdentityLimiter() *rate.Limiter
• func NewIdentityLimiter() *rate.Limiter
• func NewSendLimiter() *rate.Limiter
• func WaitForEvent(ctx context.Context, h EventHandler, ch <-chan Event, fn func(*OP) bool) error
• type Conn
  • func NewConn() *Conn
  • func NewConnWithDialer(dialer websocket.Dialer) *Conn
  • func (c *Conn) Close() error
  • func (c *Conn) Dial(ctx context.Context, addr string) (err error)
  • func (c *Conn) Listen() <-chan Event
  • func (c *Conn) Send(ctx context.Context, b []byte) error
• type Connection
• type Event
• type EventHandler
• type EventLoopHandler
• type ExtraHandler
• type ExtraHandlers
  • func (ex *ExtraHandlers) Add(check func(*OP) bool) (<-chan *OP, func())
  • func (ex *ExtraHandlers) Check(op *OP)
• type OP
  • func AssertEvent(ev Event, code OPCode, v interface{}) (*OP, error)
  • func DecodeOP(ev Event) (*OP, error)
  • func (op *OP) UnmarshalData(v interface{}) error
• type OPCode
• type PacemakerLoop
  • func (p *PacemakerLoop) Pace(ctx context.Context) error
  • func (p *PacemakerLoop) RunAsync(heartrate time.Duration, evs <-chan Event, evl EventLoopHandler, ...)
• type Websocket
  • func New(addr string) *Websocket
  • func NewCustom(conn Connection, addr string) *Websocket
  • func (ws *Websocket) Close() error
  • func (ws *Websocket) Dial(ctx context.Context) error
  • func (ws *Websocket) Listen() <-chan Event
  • func (ws *Websocket) Send(b []byte) error
  • func (ws *Websocket) SendCtx(ctx context.Context, b []byte) error

Variables

var (
	// WSTimeout is the timeout for connecting and writing to the Websocket,
	// before Gateway cancels and fails.
	WSTimeout = 30 * time.Second
	// WSBuffer is the size of the Event channel. This has to be at least 1 to
	// make space for the first Event: Ready or Resumed.
	WSBuffer = 10
	// WSError is the default error handler
	WSError = func(err error) { log.Println("Gateway error:", err) }
	// WSDebug is used for extra debug logging. This is expected to behave
	// similarly to log.Println().
	WSDebug = func(v ...interface{}) {}
)

var CloseDeadline = time.Second

CloseDeadline controls the deadline to wait for sending the Close frame.

var CopyBufferSize = 4096

CopyBufferSize is used for the initial size of the internal WS' buffer. Its size is 4KB.

var ErrEmptyPayload = errors.New("empty payload")

var ErrWebsocketClosed = errors.New("websocket is closed")

ErrWebsocketClosed is returned if the websocket is already closed.

var MaxCapUntilReset = CopyBufferSize * 4

MaxCapUntilReset determines the maximum capacity before the bytes buffer is re-allocated. It is roughly 16KB, quadruple CopyBufferSize.

Functions

func ErrBrokenConnection

func ErrBrokenConnection(err error) error

ErrBrokenConnection marks the given error as a broken connection error. This error will cause the pacemaker loop to break and return the error. The error, when stringified, will say "explicit connection break."

func HandleEvent

func HandleEvent(h EventHandler, ev Event) error

func IsBrokenConnection

func IsBrokenConnection(err error) bool

IsBrokenConnection returns true if the error is a broken connection error.

func NewDialLimiter

func NewDialLimiter() *rate.Limiter

func NewGlobalIdentityLimiter

func NewGlobalIdentityLimiter() *rate.Limiter

func NewIdentityLimiter

func NewIdentityLimiter() *rate.Limiter

func NewSendLimiter

func NewSendLimiter() *rate.Limiter

func WaitForEvent

func WaitForEvent(ctx context.Context, h EventHandler, ch <-chan Event, fn func(*OP) bool) error

WaitForEvent blocks until fn() returns true. All incoming events are handled regardless.

Types

type Conn

type Conn struct {
	Dialer websocket.Dialer
	Header http.Header
	Conn   *websocket.Conn
	// contains filtered or unexported fields
}

Conn is the default Websocket connection. It tries to compresses all payloads using zlib.

func NewConn

func NewConn() *Conn

NewConn creates a new default websocket connection with a default dialer.

func NewConnWithDialer

func NewConnWithDialer(dialer websocket.Dialer) *Conn

NewConn creates a new default websocket connection with a custom dialer.

func (*Conn) Close

func (c *Conn) Close() error

func (*Conn) Dial

func (c *Conn) Dial(ctx context.Context, addr string) (err error)

func (*Conn) Listen

func (c *Conn) Listen() <-chan Event

Listen returns an event channel if there is a connection associated with it. It returns nil if there is none.

func (*Conn) Send

func (c *Conn) Send(ctx context.Context, b []byte) error

type Connection

type Connection interface {
	// Dial dials the address (string). Context needs to be passed in for
	// timeout. This method should also be re-usable after Close is called.
	Dial(context.Context, string) error

	// Listen returns an event channel that sends over events constantly. It can
	// return nil if there isn't an ongoing connection.
	Listen() <-chan Event

	// Send allows the caller to send bytes. It does not need to clean itself
	// up on errors, as the Websocket wrapper will do that.
	Send(context.Context, []byte) error

	// Close should close the websocket connection. The underlying connection
	// may be reused, but this Connection instance will be reused with Dial. The
	// Connection must still be reusable even if Close returns an error.
	Close() error
}

Connection is an interface that abstracts around a generic Websocket driver. This connection expects the driver to handle compression by itself, including modifying the connection URL. The implementation doesn't have to be safe for concurrent use.

type Event

type Event struct {
	Data []byte

	// Error is non-nil if Data is nil.
	Error error
}

type EventHandler

type EventHandler interface {
	HandleOP(op *OP) error
}

type EventLoopHandler

type EventLoopHandler interface {
	EventHandler
	HeartbeatCtx(context.Context) error
}

TODO API

type ExtraHandler

type ExtraHandler struct {
	Check func(*OP) bool
	// contains filtered or unexported fields
}

type ExtraHandlers

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

func (*ExtraHandlers) Add

func (ex *ExtraHandlers) Add(check func(*OP) bool) (<-chan *OP, func())

func (*ExtraHandlers) Check

func (ex *ExtraHandlers) Check(op *OP)

Check runs and sends OP data. It is not thread-safe.

type OP

type OP struct {
	Code OPCode   `json:"op"`
	Data json.Raw `json:"d,omitempty"`

	// Only for Gateway Dispatch (op 0)
	Sequence  int64  `json:"s,omitempty"`
	EventName string `json:"t,omitempty"`
}

func AssertEvent

func AssertEvent(ev Event, code OPCode, v interface{}) (*OP, error)

func DecodeOP

func DecodeOP(ev Event) (*OP, error)

func (*OP) UnmarshalData

func (op *OP) UnmarshalData(v interface{}) error

type OPCode

type OPCode uint8

OPCode is a generic type for websocket OP codes.

type PacemakerLoop

type PacemakerLoop struct {
	heart.Pacemaker
	Extras   ExtraHandlers
	ErrorLog func(error)
	// contains filtered or unexported fields
}

PacemakerLoop provides an event loop with a pacemaker. A zero-value instance is a valid instance only when RunAsync is called first.

func (*PacemakerLoop) Pace

func (p *PacemakerLoop) Pace(ctx context.Context) error

Pace calls the pacemaker's Pace function.

func (*PacemakerLoop) RunAsync

func (p *PacemakerLoop) RunAsync(
	heartrate time.Duration, evs <-chan Event, evl EventLoopHandler, exit func(error))

type Websocket

type Websocket struct {

	// Timeout for connecting and writing to the Websocket, uses default
	// WSTimeout (global).
	Timeout time.Duration
	// contains filtered or unexported fields
}

Websocket is a wrapper around a websocket Conn with thread safety and rate limiting for sending and throttling.

func New

func New(addr string) *Websocket

New creates a default Websocket with the given address.

func NewCustom

func NewCustom(conn Connection, addr string) *Websocket

NewCustom creates a new undialed Websocket.

func (*Websocket) Close

func (ws *Websocket) Close() error

Close closes the websocket connection. It assumes that the Websocket is closed even when it returns an error. If the Websocket was already closed before, ErrWebsocketClosed will be returned.

func (*Websocket) Dial

func (ws *Websocket) Dial(ctx context.Context) error

Dial waits until the rate limiter allows then dials the websocket.

func (*Websocket) Listen

func (ws *Websocket) Listen() <-chan Event

Listen returns the inner event channel or nil if the Websocket connection is not alive.

func (*Websocket) Send

func (ws *Websocket) Send(b []byte) error

Send sends b over the Websocket without a timeout.

func (*Websocket) SendCtx

func (ws *Websocket) SendCtx(ctx context.Context, b []byte) error

SendCtx sends b over the Websocket with a deadline. It closes the internal Websocket if the Send method errors out.