websocket

Documentation

Index

• Constants
• Variables
• func HeadProcessor(ctx context.Context, payload string, repo identifiable[storage.Block]) (*responses.Block, error)
• func TxProcessor(ctx context.Context, payload string, repo identifiable[storage.Tx]) (*responses.Tx, error)
• type Channel
  • func NewChannel[I, M any](storageChannelName string, processor processor[I, M], repo identifiable[I], ...) *Channel[I, M]
  • func (channel *Channel[I, M]) AddClient(c client)
  • func (channel *Channel[I, M]) Close() error
  • func (channel *Channel[I, M]) RemoveClient(id uint64)
  • func (channel *Channel[I, M]) Start(ctx context.Context, factory storage.ListenerFactory)
  • func (channel *Channel[I, M]) String() string
• type Client
  • func (c *Client) ApplyFilters(msg Subscribe) error
  • func (c *Client) Close() error
  • func (c *Client) DetachFilters(msg Unsubscribe) error
  • func (c *Client) Filters() *Filters
  • func (c *Client) Id() uint64
  • func (c *Client) Notify(msg any)
  • func (c *Client) ReadMessages(ctx context.Context, ws *websocket.Conn, sub *Client, log echo.Logger)
  • func (c *Client) WriteMessages(ctx context.Context, ws *websocket.Conn, log echo.Logger)
• type Filterable
• type Filters
• type HeadFilter
  • func (hf HeadFilter) Filter(c client, msg *responses.Block) bool
• type Manager
  • func NewManager(factory storage.ListenerFactory, blockRepo storage.IBlock, txRepo storage.ITx) *Manager
  • func (manager *Manager) AddClientToChannel(channel string, client *Client)
  • func (manager *Manager) Close() error
  • func (manager *Manager) Handle(c echo.Context) error
  • func (manager *Manager) RemoveClientFromChannel(channel string, client *Client)
  • func (manager *Manager) Start(ctx context.Context)
• type Message
• type Subscribe
• type TransactionFilters
• type TxFilter
  • func (hf TxFilter) Filter(c client, msg *responses.Tx) bool
• type Unsubscribe

Constants

const (
	MethodSubscribe   = "subscribe"
	MethodUnsubscribe = "unsubscribe"
)

methods

const (
	ChannelHead = "head"
	ChannelTx   = "tx"
)

channels

Variables

var (
	ErrUnknownMethod     = errors.New("unknown method")
	ErrUnknownChannel    = errors.New("unknown channel")
	ErrUnavailableFilter = errors.New("unknown filter value")
)

Functions

func HeadProcessor

func HeadProcessor(ctx context.Context, payload string, repo identifiable[storage.Block]) (*responses.Block, error)

func TxProcessor

func TxProcessor(ctx context.Context, payload string, repo identifiable[storage.Tx]) (*responses.Tx, error)

Types

type Channel

type Channel[I, M any] struct {
	// contains filtered or unexported fields
}

func NewChannel

func NewChannel[I, M any](storageChannelName string, processor processor[I, M], repo identifiable[I], filters Filterable[M]) *Channel[I, M]

func (*Channel[I, M]) AddClient

func (channel *Channel[I, M]) AddClient(c client)

func (*Channel[I, M]) Close

func (channel *Channel[I, M]) Close() error

func (*Channel[I, M]) RemoveClient

func (channel *Channel[I, M]) RemoveClient(id uint64)

func (*Channel[I, M]) Start

func (channel *Channel[I, M]) Start(ctx context.Context, factory storage.ListenerFactory)

func (*Channel[I, M]) String

func (channel *Channel[I, M]) String() string

type Client

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

func (*Client) ApplyFilters

func (c *Client) ApplyFilters(msg Subscribe) error

func (*Client) Close

func (c *Client) Close() error

func (*Client) DetachFilters

func (c *Client) DetachFilters(msg Unsubscribe) error

func (*Client) Filters

func (c *Client) Filters() *Filters

func (*Client) Id

func (c *Client) Id() uint64

func (*Client) Notify

func (c *Client) Notify(msg any)

func (*Client) ReadMessages

func (c *Client) ReadMessages(ctx context.Context, ws *websocket.Conn, sub *Client, log echo.Logger)

func (*Client) WriteMessages

func (c *Client) WriteMessages(ctx context.Context, ws *websocket.Conn, log echo.Logger)

type Filterable

type Filterable[M any] interface {
	Filter(c client, msg M) bool
}

type Filters

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

type HeadFilter

type HeadFilter struct{}

func (HeadFilter) Filter

func (hf HeadFilter) Filter(c client, msg *responses.Block) bool

type Manager

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

func NewManager

func NewManager(factory storage.ListenerFactory, blockRepo storage.IBlock, txRepo storage.ITx) *Manager

func (*Manager) AddClientToChannel

func (manager *Manager) AddClientToChannel(channel string, client *Client)

func (*Manager) Close

func (manager *Manager) Close() error

func (*Manager) Handle

func (manager *Manager) Handle(c echo.Context) error

Handle godoc

@Summary				Websocket API
@Description.markdown	websocket
@Tags					websocket
@ID						websocket
@Produce				json
@Router					/v1/ws [get]

func (*Manager) RemoveClientFromChannel

func (manager *Manager) RemoveClientFromChannel(channel string, client *Client)

func (*Manager) Start

func (manager *Manager) Start(ctx context.Context)

type Message

type Message struct {
	Method string          `json:"method" validate:"required,oneof=subscribe,unsubscribe"`
	Body   json.RawMessage `json:"body"   validate:"required"`
}

type Subscribe

type Subscribe struct {
	Channel string          `json:"channel" validate:"required,oneof=head tx"`
	Filters json.RawMessage `json:"filters" validate:"required"`
}

type TransactionFilters

type TransactionFilters struct {
	Status   []string `json:"status,omitempty"`
	Messages []string `json:"msg_type,omitempty"`
}

type TxFilter

type TxFilter struct{}

func (TxFilter) Filter

func (hf TxFilter) Filter(c client, msg *responses.Tx) bool

type Unsubscribe

type Unsubscribe struct {
	Channel string `json:"channel" validate:"required,oneof=head tx"`
}