ssepub

Documentation

Index

• Constants
• type HeaderModifications
• type MessageHandler
• type SSEClient
  • func NewSSEClient(id nxid.ID, maxRetries int, handler MessageHandler, req *http.Request, ...) *SSEClient
  • func (sc *SSEClient) Close() error
  • func (sc *SSEClient) Send(method string, msg *sabuhp.Message, timeout time.Duration) error
  • func (sc *SSEClient) Wait()
• type SSEHub
  • func NewSSEHub(ctx context.Context, maxRetries int, client *http.Client, codec sabuhp.Codec, ...) *SSEHub
  • func (se *SSEHub) For(handler MessageHandler, id nxid.ID, method string, route string, ...) (*SSEClient, error)
  • func (se *SSEHub) Get(handler MessageHandler, id nxid.ID, route string, lastEventIds ...string) (*SSEClient, error)
• type SSEServer
  • func ManagedSSEServer(ctx context.Context, logger sabuhp.Logger, manager *managers.Manager, ...) *SSEServer
  • func (sse *SSEServer) Handle(w http.ResponseWriter, r *http.Request, p sabuhp.Params)
  • func (sse *SSEServer) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type SSESocket
  • func NewSSESocket(clientId string, ctx context.Context, r *http.Request, w http.ResponseWriter, ...) *SSESocket
  • func (se *SSESocket) ID() nxid.ID
  • func (se *SSESocket) LocalAddr() net.Addr
  • func (se *SSESocket) RemoteAddr() net.Addr
  • func (se *SSESocket) Send(msg []byte, meta sabuhp.MessageMeta, timeout time.Duration) error
  • func (se *SSESocket) SendRead(msg *sabuhp.Message, timeout time.Duration) error
  • func (se *SSESocket) SendWriter(msgWriter io.WriterTo, meta sabuhp.MessageMeta, timeout time.Duration) sabuhp.ErrorWaiter
  • func (se *SSESocket) Start() error
  • func (se *SSESocket) Stat() sabuhp.SocketStat
  • func (se *SSESocket) Stop()
  • func (se *SSESocket) Wait()

Constants

const (
	MinClientIdLength          = 7
	ClientIdentificationHeader = "X-SSE-Client-Id"
	LastEventIdListHeader      = "X-SSE-Last-Event-Ids"
)

Types

type HeaderModifications

type HeaderModifications func(header http.Header)

type MessageHandler

type MessageHandler func(message *sabuhp.Message, socket *SSEClient) error

type SSEClient

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

func NewSSEClient

func NewSSEClient(
	id nxid.ID,
	maxRetries int,
	handler MessageHandler,
	req *http.Request,
	res *http.Response,
	retryFn sabuhp.RetryFunc,
	codec sabuhp.Codec,
	logger sabuhp.Logger,
	reqClient *http.Client,
) *SSEClient

func (*SSEClient) Close

func (sc *SSEClient) Close() error

Close closes client's request and response cycle and waits till managing goroutine is closed.

func (*SSEClient) Send

func (sc *SSEClient) Send(method string, msg *sabuhp.Message, timeout time.Duration) error

func (*SSEClient) Wait

func (sc *SSEClient) Wait()

Wait blocks till client and it's managing goroutine closes.

type SSEHub

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

func NewSSEHub

func NewSSEHub(
	ctx context.Context,
	maxRetries int,
	client *http.Client,
	codec sabuhp.Codec,
	logging sabuhp.Logger,
	retryFn sabuhp.RetryFunc,
) *SSEHub

func (*SSEHub) For

func (se *SSEHub) For(
	handler MessageHandler,
	id nxid.ID,
	method string,
	route string,
	lastEventIds ...string,
) (*SSEClient, error)

func (*SSEHub) Get

func (se *SSEHub) Get(handler MessageHandler, id nxid.ID, route string, lastEventIds ...string) (*SSEClient, error)

type SSEServer

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

func ManagedSSEServer

func ManagedSSEServer(
	ctx context.Context,
	logger sabuhp.Logger,
	manager *managers.Manager,
	optionalHeaders HeaderModifications,
	codec sabuhp.Codec,
) *SSEServer

func (*SSEServer) Handle

func (sse *SSEServer) Handle(w http.ResponseWriter, r *http.Request, p sabuhp.Params)

func (*SSEServer) ServeHTTP

func (sse *SSEServer) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements the http.Handler interface.

It collects all values from http.Request.ParseForm() as params map and calls them with SSEServer.Handle.

type SSESocket

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

func NewSSESocket

func NewSSESocket(
	clientId string,
	ctx context.Context,
	r *http.Request,
	w http.ResponseWriter,
	params sabuhp.Params,
	codec sabuhp.Codec,
	logger sabuhp.Logger,
	manager *managers.Manager,
	optionalHeaders HeaderModifications,
) *SSESocket

func (*SSESocket) ID

func (se *SSESocket) ID() nxid.ID

func (*SSESocket) LocalAddr

func (se *SSESocket) LocalAddr() net.Addr

func (*SSESocket) RemoteAddr

func (se *SSESocket) RemoteAddr() net.Addr

func (*SSESocket) Send

func (se *SSESocket) Send(msg []byte, meta sabuhp.MessageMeta, timeout time.Duration) error

func (*SSESocket) SendRead

func (se *SSESocket) SendRead(msg *sabuhp.Message, timeout time.Duration) error

func (*SSESocket) SendWriter

func (se *SSESocket) SendWriter(msgWriter io.WriterTo, meta sabuhp.MessageMeta, timeout time.Duration) sabuhp.ErrorWaiter

func (*SSESocket) Start

func (se *SSESocket) Start() error

func (*SSESocket) Stat

func (se *SSESocket) Stat() sabuhp.SocketStat

func (*SSESocket) Stop

func (se *SSESocket) Stop()

func (*SSESocket) Wait

func (se *SSESocket) Wait()