events

Documentation

Index

• Constants
• type ResetRequest
• type StreamInfo
  • func (spec *StreamInfo) GetID() string
• type SubscriptionManager
  • func NewSubscriptionManager(config *conf.EventstreamConf, rpc client.RPCClient, ...) SubscriptionManager

Constants

const (
	// when multiple websocket client are connected for the same topic, send events to all connected clients
	DistributionModeBroadcast = "broadcast"
	// when multiple websocket client are connected for the same topic, send each event to only one of the connected clients
	DistributionModeWLD = "workloadDistribution"
	// send events via a webhook endpoint
	EventStreamTypeWebhook = "webhook"
	// send events via a websocket connection
	EventStreamTypeWebsocket = "websocket"
	// FromBlockNewest is the special string that means subscribe from the current block
	FromBlockNewest = "newest"
	// ErrorHandlingBlock blocks the event stream until the handler can accept the event
	ErrorHandlingBlock = "block"
	// ErrorHandlingSkip processes up to the retry behavior on the stream, then skips to the next event
	ErrorHandlingSkip = "skip"
	// MaxBatchSize is the maximum that a user can specific for their batch size
	MaxBatchSize = 1000

	DefaultExponentialBackoffInitial = time.Duration(1) * time.Second
	DefaultExponentialBackoffFactor  = float64(2.0)
	DefaultTimestampCacheSize        = 1000
	DefaultBlockedRetryDelaySec      = 30
	DefaultBatchTimeoutMS            = 5000
	DefaultErrorHandling             = ErrorHandlingSkip
)

const (
	// SubPathPrefix is the path prefix for subscriptions
	SubPathPrefix = "/subscriptions"
	// StreamPathPrefix is the path prefix for event streams
	StreamPathPrefix = "/eventstreams"
)

Types

type ResetRequest

type ResetRequest struct {
	InitialBlock string `json:"initialBlock"`
}

type StreamInfo

type StreamInfo struct {
	eventsapi.TimeSorted
	ID                   string               `json:"id"`
	Name                 string               `json:"name,omitempty"`
	Path                 string               `json:"path,omitempty"`
	Suspended            *bool                `json:"suspended,omitempty"`
	Type                 string               `json:"type"`
	BatchSize            uint64               `json:"batchSize,omitempty"`
	BatchTimeoutMS       uint64               `json:"batchTimeoutMS,omitempty"`
	ErrorHandling        string               `json:"errorHandling,omitempty"`
	RetryTimeoutSec      uint64               `json:"retryTimeoutSec,omitempty"`
	BlockedRetryDelaySec uint64               `json:"blockedRetryDelaySec,omitempty"`
	Webhook              *webhookActionInfo   `json:"webhook,omitempty"`
	WebSocket            *webSocketActionInfo `json:"websocket,omitempty"`
	Timestamps           *bool                `json:"timestamps,omitempty"` // Include block timestamps in the events generated
	TimestampCacheSize   int                  `json:"timestampCacheSize,omitempty"`
}

StreamInfo configures the stream to perform an action for each event

func (*StreamInfo) GetID

func (spec *StreamInfo) GetID() string

GetID returns the ID (for sorting)

type SubscriptionManager

type SubscriptionManager interface {
	Init(mocked ...kvstore.KVStore) error
	AddStream(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*StreamInfo, *restutil.RestError)
	Streams(res http.ResponseWriter, req *http.Request, params httprouter.Params) []*StreamInfo
	StreamByID(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*StreamInfo, *restutil.RestError)
	UpdateStream(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*StreamInfo, *restutil.RestError)
	SuspendStream(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*map[string]string, *restutil.RestError)
	ResumeStream(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*map[string]string, *restutil.RestError)
	DeleteStream(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*map[string]string, *restutil.RestError)
	AddSubscription(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*eventsapi.SubscriptionInfo, *restutil.RestError)
	Subscriptions(res http.ResponseWriter, req *http.Request, params httprouter.Params) []*eventsapi.SubscriptionInfo
	SubscriptionByID(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*eventsapi.SubscriptionInfo, *restutil.RestError)
	ResetSubscription(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*map[string]string, *restutil.RestError)
	DeleteSubscription(res http.ResponseWriter, req *http.Request, params httprouter.Params) (*map[string]string, *restutil.RestError)
	Close()
}

SubscriptionManager provides REST APIs for managing events

func NewSubscriptionManager

func NewSubscriptionManager(config *conf.EventstreamConf, rpc client.RPCClient, wsChannels ws.WebSocketChannels) SubscriptionManager

NewSubscriptionManager constructor