broker

Documentation

Index

• type Broker
• type ChannelBroker
  • func NewChannelBroker() *ChannelBroker
  • func (b *ChannelBroker) Publish(ctx context.Context, topic string, data []byte) error
  • func (b *ChannelBroker) ReadRequest(ctx context.Context) (json.RawMessage, func(json.RawMessage) error, error)
  • func (b *ChannelBroker) Subscribe(ctx context.Context, topic string) (Subscription, error)
  • func (b *ChannelBroker) WriteRequest(ctx context.Context, topic string, msg json.RawMessage) error
• type Client
  • func NewClient(spoke ClientSpoke) *Client
  • func (c *Client) BatchCall(ctx context.Context, b ...*codec.BatchElem) error
  • func (c *Client) Close() error
  • func (c *Client) Closed() <-chan struct{}
  • func (c *Client) Do(ctx context.Context, result any, method string, params any) error
  • func (c *Client) Mount(h codec.Middleware)
  • func (c *Client) Notify(ctx context.Context, method string, params any) error
  • func (c *Client) SetHandlerPeer(pi codec.PeerInfo)
  • func (c *Client) SetHeader(key string, value string)
• type ClientSpoke
• type Codec
  • func NewCodec(req json.RawMessage, replier func(json.RawMessage) error) *Codec
  • func (c *Codec) Close() error
  • func (c *Codec) Closed() <-chan struct{}
  • func (c *Codec) Flush() (err error)
  • func (c *Codec) PeerInfo() codec.PeerInfo
  • func (c *Codec) ReadBatch(ctx context.Context) ([]*codec.Message, bool, error)
  • func (c *Codec) RemoteAddr() string
  • func (c *Codec) Write(p []byte) (n int, err error)
• type Server
  • func (s *Server) ServeSpoke(ctx context.Context, stream ServerSpoke)
• type ServerSpoke
• type Subscription

Types

type Broker

type Broker interface {
	ServerSpoke
	ClientSpoke
}

type ChannelBroker

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

func NewChannelBroker

func NewChannelBroker() *ChannelBroker

func (*ChannelBroker) Publish

func (b *ChannelBroker) Publish(ctx context.Context, topic string, data []byte) error

func (*ChannelBroker) ReadRequest

func (b *ChannelBroker) ReadRequest(ctx context.Context) (json.RawMessage, func(json.RawMessage) error, error)

func (*ChannelBroker) Subscribe

func (b *ChannelBroker) Subscribe(ctx context.Context, topic string) (Subscription, error)

func (*ChannelBroker) WriteRequest

func (b *ChannelBroker) WriteRequest(ctx context.Context, topic string, msg json.RawMessage) error

type Client

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

func NewClient

func NewClient(spoke ClientSpoke) *Client

func (*Client) BatchCall

func (c *Client) BatchCall(ctx context.Context, b ...*codec.BatchElem) error

func (*Client) Close

func (c *Client) Close() error

func (*Client) Closed

func (c *Client) Closed() <-chan struct{}

func (*Client) Do

func (c *Client) Do(ctx context.Context, result any, method string, params any) error

func (*Client) Mount

func (c *Client) Mount(h codec.Middleware)

func (*Client) Notify

func (c *Client) Notify(ctx context.Context, method string, params any) error

func (*Client) SetHandlerPeer

func (c *Client) SetHandlerPeer(pi codec.PeerInfo)

func (*Client) SetHeader

func (c *Client) SetHeader(key string, value string)

type ClientSpoke

type ClientSpoke interface {
	WriteRequest(ctx context.Context, clientId string, msg json.RawMessage) error
	Subscribe(ctx context.Context, clientId string) (Subscription, error)
}

type Codec

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

func NewCodec

func NewCodec(req json.RawMessage, replier func(json.RawMessage) error) *Codec

func (*Codec) Close

func (c *Codec) Close() error

closes the connection

func (*Codec) Closed

func (c *Codec) Closed() <-chan struct{}

Closed returns a channel which is closed when the connection is closed.

func (*Codec) Flush

func (c *Codec) Flush() (err error)

func (*Codec) PeerInfo

func (c *Codec) PeerInfo() codec.PeerInfo

gets the peer info

func (*Codec) ReadBatch

func (c *Codec) ReadBatch(ctx context.Context) ([]*codec.Message, bool, error)

func (*Codec) RemoteAddr

func (c *Codec) RemoteAddr() string

RemoteAddr returns the peer address of the connection.

func (*Codec) Write

func (c *Codec) Write(p []byte) (n int, err error)

type Server

type Server struct {
	Server *server.Server
}

func (*Server) ServeSpoke

func (s *Server) ServeSpoke(ctx context.Context, stream ServerSpoke)

type ServerSpoke

type ServerSpoke interface {
	ReadRequest(ctx context.Context) (json.RawMessage, func(json.RawMessage) error, error)
}

type Subscription

type Subscription interface {
	// channel that will close when done or error
	Listen() <-chan json.RawMessage
	// should close the channel and also stop listening
	Close() error
	// this hold errors
	Err() error
}