stream

Documentation

Index

• Variables
• type Message
• type MsgProcessor
• type Retryer
• type Stream
  • func New(l log.Logger, processor MsgProcessor, retryer Retryer, endpoint string) *Stream
  • func (s *Stream) ListenAndProcess(ctx context.Context) (<-chan error, error)
• type WSDialerWithRetries
  • func NewWSDialerWithRetries(dialer *websocket.Dialer, l log.Logger, r Retryer) *WSDialerWithRetries
  • func (ws *WSDialerWithRetries) Dial(ctx context.Context, urlStr string, requestHeader http.Header) (*websocket.Conn, *http.Response, error)

Variables

var ReadTimeout = 5

ReadTimeout specifies the time, in seconds, the streams wait for reading the next message.

Types

type Message

type Message struct {
	CheckID string `json:"check_id,omitempty"`
	AgentID string `json:"agent_id,omitempty"`
	ScanID  string `json:"scan_id,omitempty"`
	Action  string `json:"action"`
}

Message describes a stream message.

type MsgProcessor

type MsgProcessor interface {
	AbortCheck(ID string)
}

MsgProcessor defines the function that the Stream will call when an abort message has been received.

type Retryer

type Retryer interface {
	WithRetries(op string, exec func() error) error
}

Retryer represents the functions used by the Stream for retrying when connecting to the stream.

type Stream

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

Stream reads messages from the stream server and process them using a given message processor.

func New

func New(l log.Logger, processor MsgProcessor, retryer Retryer, endpoint string) *Stream

New creates a new stream that will use the given processor to process the messages received by the Stream.

func (*Stream) ListenAndProcess

func (s *Stream) ListenAndProcess(ctx context.Context) (<-chan error, error)

ListenAndProcess start listening for new messages from a stream.

type WSDialerWithRetries

type WSDialerWithRetries struct {
	*websocket.Dialer
	// contains filtered or unexported fields
}

WSDialerWithRetries provides retries with backoff and jitter when initiating a connection to a websocket.

func NewWSDialerWithRetries

func NewWSDialerWithRetries(dialer *websocket.Dialer, l log.Logger, r Retryer) *WSDialerWithRetries

NewWSDialerWithRetries creates a WSDialer with the given retries parameters.

func (*WSDialerWithRetries) Dial

func (ws *WSDialerWithRetries) Dial(ctx context.Context, urlStr string, requestHeader http.Header) (*websocket.Conn, *http.Response, error)

Dial wraps the dial function of the websocket dialer adding retries functionality.