events

Documentation

Index

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

Constants

const (
	// 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  is the initial delay for backoff retry
	DefaultExponentialBackoffInitial = time.Duration(1) * time.Second
	// DefaultExponentialBackoffFactor is the factor we use between retries
	DefaultExponentialBackoffFactor = float64(2.0)
	// DefaultTimestampCacheSize is the number of entries we will hold in a LRU cache for block timestamps
	DefaultTimestampCacheSize = 1000
)

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

Types

type DistributionMode

type DistributionMode string

const (
	DistributionModeBroadcast DistributionMode = "broadcast"
	DistributionModeWLD       DistributionMode = "workloadDistribution"
)

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"`
	Suspended            bool                 `json:"suspended"`
	Type                 string               `json:"type,omitempty"`
	BatchSize            uint64               `json:"batchSize,omitempty"`
	BatchTimeoutMS       uint64               `json:"batchTimeoutMS,omitempty"`
	ErrorHandling        string               `json:"errorHandling,omitempty"`
	RetryTimeoutSec      uint64               `json:"retryTimeoutSec,omitempty"`
	BlockedRetryDelaySec uint64               `json:"blockedReryDelaySec,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() 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