websockets

README

Websocket for FNS

Install

go get github.com/aacfactory/fns-contrib/http/handlers/websockets

Usage

Make sure tls is used.

app := fns.New(
    fns.Handlers(websockets.Websocket()),
)

Setup config

http:
  handlers:
    - websockets:
        maxConnections: 1024
        handshakeTimeout: "1s"
        readTimeout: "10s"
        readBufferSize: "4MB"
        writeTimeout: "60s"
        writeBufferSize: "4MB"
        enableCompression: false
        maxRequestMessageSize: "4KB"

Enable sub protocol handler, such as MQTT.

app := fns.New(
    fns.Handlers(websockets.Websocket(subs...)),
)

Get connection id in function

connId := websockets.ConnectionId(ctx)

Send message to client

err := websockets.Send(ctx, connId, payload)

Documentation

Index

• Variables
• func ConnectionId(ctx context.Context) (id string)
• func Send(ctx context.Context, connectionId string, payload interface{}) (err error)
• func Websocket(subs ...SubProtocolHandler) (handler service.HttpHandler)
• type Config
• type Connection
• type MessageType
• type Request
  • func (request *Request) DeviceId() (id string)
  • func (request *Request) DeviceIp() (id string)
  • func (request *Request) Validate() (err error)
• type Response
  • func Failed(err error) (resp *Response)
  • func Succeed(payload interface{}) (resp *Response)
  • func (resp *Response) Encode() (p []byte)
• type Service
  • func (svc *Service) Build(options service.Options) (err error)
  • func (svc *Service) Components() (components map[string]service.Component)
  • func (svc *Service) Document() (doc service.Document)
  • func (svc *Service) Handle(ctx context.Context, fn string, argument service.Argument) (v interface{}, err errors.CodeError)
• type SubProtocolHandler
• type SubProtocolHandlerOptions
• type SubProtocolHandlerTask
  • func (t *SubProtocolHandlerTask) Execute(ctx context.Context)
• type Task
  • func (t *Task) Execute(ctx context.Context)
• type WebsocketConnection
  • func (conn *WebsocketConnection) NextReader() (messageType MessageType, r io.Reader, err error)
  • func (conn *WebsocketConnection) NextWriter(messageType MessageType) (w io.WriteCloser, err error)
  • func (conn *WebsocketConnection) ReadMessage() (messageType MessageType, p []byte, err error)
  • func (conn *WebsocketConnection) WriteControl(messageType MessageType, data []byte, deadline time.Time) error
  • func (conn *WebsocketConnection) WriteMessage(messageType MessageType, data []byte) (err error)

Variables

var (
	ErrRequestMessageIsTooLarge = fmt.Errorf("message is too large")
)

Functions

func ConnectionId

func ConnectionId(ctx context.Context) (id string)

func Send

func Send(ctx context.Context, connectionId string, payload interface{}) (err error)

func Websocket

func Websocket(subs ...SubProtocolHandler) (handler service.HttpHandler)

Types

type Config

type Config struct {
	MaxConnections        int    `json:"maxConnections"`
	HandshakeTimeout      string `json:"handshakeTimeout"`
	ReadTimeout           string `json:"readTimeout"`
	ReadBufferSize        string `json:"readBufferSize"`
	WriteTimeout          string `json:"writeTimeout"`
	WriteBufferSize       string `json:"writeBufferSize"`
	EnableCompression     bool   `json:"enableCompression"`
	MaxRequestMessageSize string `json:"maxRequestMessageSize"`
}

type Connection

type Connection interface {
	ReadMessage() (messageType MessageType, p []byte, err error)
	WriteMessage(messageType MessageType, data []byte) (err error)
	WriteControl(messageType MessageType, data []byte, deadline time.Time) error
	NextWriter(messageType MessageType) (io.WriteCloser, error)
	SetWriteDeadline(t time.Time) error
	NextReader() (messageType MessageType, r io.Reader, err error)
	SetReadDeadline(t time.Time) error
	SetReadLimit(limit int64)
	CloseHandler() func(code int, text string) error
	SetCloseHandler(h func(code int, text string) error)
	PingHandler() func(appData string) error
	SetPingHandler(h func(appData string) error)
	PongHandler() func(appData string) error
	SetPongHandler(h func(appData string) error)
	UnderlyingConn() net.Conn
	EnableWriteCompression(enable bool)
	SetCompressionLevel(level int) error
	Subprotocol() (protocol string)
	Close() (err error)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
}

type MessageType

type MessageType int

type Request

type Request struct {
	Service string          `json:"service"`
	Fn      string          `json:"fn"`
	Header  http.Header     `json:"header"`
	Payload json.RawMessage `json:"payload"`
}

func (*Request) DeviceId

func (request *Request) DeviceId() (id string)

func (*Request) DeviceIp

func (request *Request) DeviceIp() (id string)

func (*Request) Validate

func (request *Request) Validate() (err error)

type Response

type Response struct {
	Succeed bool        `json:"succeed"`
	Payload interface{} `json:"result"`
}

func Failed

func Failed(err error) (resp *Response)

func Succeed

func Succeed(payload interface{}) (resp *Response)

func (*Response) Encode

func (resp *Response) Encode() (p []byte)

type Service

type Service struct {
	service.Abstract
	// contains filtered or unexported fields
}

func (*Service) Build

func (svc *Service) Build(options service.Options) (err error)

func (*Service) Components

func (svc *Service) Components() (components map[string]service.Component)

func (*Service) Document

func (svc *Service) Document() (doc service.Document)

func (*Service) Handle

func (svc *Service) Handle(ctx context.Context, fn string, argument service.Argument) (v interface{}, err errors.CodeError)

type SubProtocolHandler

type SubProtocolHandler interface {
	Name() (name string)
	Build(options SubProtocolHandlerOptions) (err error)
	Handle(ctx context.Context, conn Connection)
	Service() (service service.Service)
	Close() (err error)
}

type SubProtocolHandlerOptions

type SubProtocolHandlerOptions struct {
	AppId                 string
	AppName               string
	AppVersion            versions.Version
	Log                   logs.Logger
	Config                configures.Config
	ReadTimeout           time.Duration
	WriteTimeout          time.Duration
	MaxRequestMessageSize int64
	Discovery             service.EndpointDiscovery `json:"-"`
}

type SubProtocolHandlerTask

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

func (*SubProtocolHandlerTask) Execute

func (t *SubProtocolHandlerTask) Execute(ctx context.Context)

type Task

type Task struct {
	*workers.AbstractLongTask
	// contains filtered or unexported fields
}

func (*Task) Execute

func (t *Task) Execute(ctx context.Context)

type WebsocketConnection

type WebsocketConnection struct {
	*websocket.Conn
}

func (*WebsocketConnection) NextReader

func (conn *WebsocketConnection) NextReader() (messageType MessageType, r io.Reader, err error)

func (*WebsocketConnection) NextWriter

func (conn *WebsocketConnection) NextWriter(messageType MessageType) (w io.WriteCloser, err error)

func (*WebsocketConnection) ReadMessage

func (conn *WebsocketConnection) ReadMessage() (messageType MessageType, p []byte, err error)

func (*WebsocketConnection) WriteControl

func (conn *WebsocketConnection) WriteControl(messageType MessageType, data []byte, deadline time.Time) error

func (*WebsocketConnection) WriteMessage

func (conn *WebsocketConnection) WriteMessage(messageType MessageType, data []byte) (err error)