wsutil

Documentation

Overview

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

Index

• Constants
• Variables
• func HandleEvent(h EventHandler, ev Event) error
• func InjectValues(rawurl string, values url.Values) string
• 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 NewConnWithDriver(driver json.Driver) *Conn
  • func (c *Conn) Close() (err error)
  • func (c *Conn) Dial(ctx context.Context, addr string) 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 NewLoop(heartrate time.Duration, evs <-chan Event, evl EventLoopHandler) *PacemakerLoop
  • func (p *PacemakerLoop) Echo()
  • func (p *PacemakerLoop) Pace(ctx context.Context) error
  • func (p *PacemakerLoop) RunAsync(exit func(error))
  • func (p *PacemakerLoop) Stop()
  • func (p *PacemakerLoop) Stopped() bool
• 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

Constants

const CopyBufferSize = 2048

Variables

var (
	// WSTimeout is the timeout for connecting and writing to the Websocket,
	// before Gateway cancels and fails.
	WSTimeout = 5 * time.Minute
	// 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) }
	// WSExtraReadTimeout is the duration to be added to Hello, as a read
	// timeout for the websocket.
	WSExtraReadTimeout = time.Second
	// 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 ErrEmptyPayload = errors.New("empty payload")

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

ErrWebsocketClosed is returned if the websocket is already closed.

Functions

func HandleEvent

func HandleEvent(h EventHandler, ev Event) error

func InjectValues

func InjectValues(rawurl string, values url.Values) string

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 {
	Conn *websocket.Conn
	json.Driver
	// contains filtered or unexported fields
}

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

func NewConn

func NewConn() *Conn

func NewConnWithDriver

func NewConnWithDriver(driver json.Driver) *Conn

func (*Conn) Close

func (c *Conn) Close() (err error)

func (*Conn) Dial

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

func (*Conn) Listen

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

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 sends over events constantly. Error will be non-nil if Data is
	// nil, so check for Error first.
	Listen() <-chan Event

	// Send allows the caller to send bytes. Thread safety is a requirement.
	Send(context.Context, []byte) error

	// Close should close the websocket connection. The connection will not be
	// reused.
	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.

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 {
	Extras ExtraHandlers

	ErrorLog func(error)
	// contains filtered or unexported fields
}

PacemakerLoop provides an event loop with a pacemaker.

func NewLoop

func NewLoop(heartrate time.Duration, evs <-chan Event, evl EventLoopHandler) *PacemakerLoop

func (*PacemakerLoop) Echo

func (p *PacemakerLoop) Echo()

Echo calls the pacemaker's Echo function.

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(exit func(error))

func (*PacemakerLoop) Stop

func (p *PacemakerLoop) Stop()

Stop calls the pacemaker's Stop function.

func (*PacemakerLoop) Stopped

func (p *PacemakerLoop) Stopped() bool

type Websocket

type Websocket struct {
	Conn Connection
	Addr string

	// Timeout for connecting and writing to the Websocket, uses default
	// WSTimeout (global).
	Timeout time.Duration

	SendLimiter *rate.Limiter
	DialLimiter *rate.Limiter
}

func New

func New(addr string) *Websocket

func NewCustom

func NewCustom(conn Connection, addr string) *Websocket

NewCustom creates a new undialed Websocket.

func (*Websocket) Close

func (ws *Websocket) Close() error

func (*Websocket) Dial

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

func (*Websocket) Listen

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

func (*Websocket) Send

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

func (*Websocket) SendCtx

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