nats

Documentation

Index

• type GlobalEvents
  • func NewGlobalEvents(kv jetstream.KeyValue, l *log.Logger) (*GlobalEvents, error)
  • func (g *GlobalEvents) CurrentLiveEvents() (lookup map[string]*containerv1.EventContainer, err error)
  • func (g *GlobalEvents) RegisterEvent(e *containerv1.EventContainer)
  • func (g *GlobalEvents) UnregisterEvent(eventKey string)
• type NatsProxy
  • func NewNatsProxy(conn *nats.Conn, opts ...Option) (*NatsProxy, error)
  • func (n *NatsProxy) Close()
  • func (n *NatsProxy) DeleteEventCallback(eventKey string)
  • func (n *NatsProxy) GetEvent(selector *commonv1.EventSelector) (*containerv1.EventContainer, error)
  • func (n *NatsProxy) HistorySnapshotData(sel *commonv1.EventSelector) []*analysisv1.SnapshotData
  • func (n *NatsProxy) LiveEvents() []*containerv1.EventContainer
  • func (n *NatsProxy) PublishAnalysisData(eventKey string, a *analysisv1.Analysis) error
  • func (n *NatsProxy) PublishDriverData(req *racestatev1.PublishDriverDataRequest) error
  • func (n *NatsProxy) PublishEventRegistered(epd *utils.EventProcessingData) error
  • func (n *NatsProxy) PublishEventUnregistered(eventKey string) error
  • func (n *NatsProxy) PublishRaceStateData(req *racestatev1.PublishStateRequest) error
  • func (n *NatsProxy) PublishSpeedmapData(req *racestatev1.PublishSpeedmapRequest) error
  • func (n *NatsProxy) SetOnUnregisterCB(cb func(sel *commonv1.EventSelector))
  • func (n *NatsProxy) SubscribeAnalysisData(sel *commonv1.EventSelector) (d <-chan *analysisv1.Analysis, q chan<- struct{}, err error)
  • func (n *NatsProxy) SubscribeDriverData(sel *commonv1.EventSelector) (d <-chan *livedatav1.LiveDriverDataResponse, q chan<- struct{}, err error)
  • func (n *NatsProxy) SubscribeRaceStateData(sel *commonv1.EventSelector) (<-chan *racestatev1.PublishStateRequest, chan<- struct{}, error)
  • func (n *NatsProxy) SubscribeSnapshotData(sel *commonv1.EventSelector) (d <-chan *analysisv1.SnapshotData, q chan<- struct{}, err error)
  • func (n *NatsProxy) SubscribeSpeedmapData(sel *commonv1.EventSelector) (<-chan *racestatev1.PublishSpeedmapRequest, chan<- struct{}, error)
• type Option
  • func WithContext(ctx context.Context) Option
  • func WithLogger(l *log.Logger) Option

Types

type GlobalEvents

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

GlobalEvents is a global event store that is used to keep track of all events this data is used during initialization of the NatsProxy. the purpse is to enable instances that are started later to be able to deliver data as well.

func NewGlobalEvents

func NewGlobalEvents(kv jetstream.KeyValue, l *log.Logger) (*GlobalEvents, error)

func (*GlobalEvents) CurrentLiveEvents

func (g *GlobalEvents) CurrentLiveEvents() (
	lookup map[string]*containerv1.EventContainer,
	err error,
)

func (*GlobalEvents) RegisterEvent

func (g *GlobalEvents) RegisterEvent(e *containerv1.EventContainer)

called when this instance is recording an event

func (*GlobalEvents) UnregisterEvent

func (g *GlobalEvents) UnregisterEvent(eventKey string)

called on the recording instance when it is no longer recording an event

type NatsProxy

type NatsProxy struct {
	proxy.EmptyProxy
	// contains filtered or unexported fields
}

func NewNatsProxy

func NewNatsProxy(conn *nats.Conn, opts ...Option) (*NatsProxy, error)

func (*NatsProxy) Close

func (n *NatsProxy) Close()

func (*NatsProxy) DeleteEventCallback

func (n *NatsProxy) DeleteEventCallback(eventKey string)

this method is called when the watchdog detects a stale event and deletes it

func (*NatsProxy) GetEvent

func (n *NatsProxy) GetEvent(selector *commonv1.EventSelector) (
	*containerv1.EventContainer, error,
)

func (*NatsProxy) HistorySnapshotData

func (n *NatsProxy) HistorySnapshotData(sel *commonv1.EventSelector) []*analysisv1.SnapshotData

func (*NatsProxy) LiveEvents

func (n *NatsProxy) LiveEvents() []*containerv1.EventContainer

func (*NatsProxy) PublishAnalysisData

func (n *NatsProxy) PublishAnalysisData(eventKey string, a *analysisv1.Analysis) error

func (*NatsProxy) PublishDriverData

func (n *NatsProxy) PublishDriverData(req *racestatev1.PublishDriverDataRequest) error

func (*NatsProxy) PublishEventRegistered

func (n *NatsProxy) PublishEventRegistered(epd *utils.EventProcessingData) error

func (*NatsProxy) PublishEventUnregistered

func (n *NatsProxy) PublishEventUnregistered(eventKey string) error

func (*NatsProxy) PublishRaceStateData

func (n *NatsProxy) PublishRaceStateData(req *racestatev1.PublishStateRequest) error

func (*NatsProxy) PublishSpeedmapData

func (n *NatsProxy) PublishSpeedmapData(req *racestatev1.PublishSpeedmapRequest) error

func (*NatsProxy) SetOnUnregisterCB

func (n *NatsProxy) SetOnUnregisterCB(cb func(sel *commonv1.EventSelector))

func (*NatsProxy) SubscribeAnalysisData

func (n *NatsProxy) SubscribeAnalysisData(sel *commonv1.EventSelector) (
	d <-chan *analysisv1.Analysis,
	q chan<- struct{},
	err error,
)

func (*NatsProxy) SubscribeDriverData

func (n *NatsProxy) SubscribeDriverData(sel *commonv1.EventSelector) (
	d <-chan *livedatav1.LiveDriverDataResponse,
	q chan<- struct{},
	err error,
)

func (*NatsProxy) SubscribeRaceStateData

func (n *NatsProxy) SubscribeRaceStateData(sel *commonv1.EventSelector) (
	<-chan *racestatev1.PublishStateRequest,
	chan<- struct{},
	error,
)

func (*NatsProxy) SubscribeSnapshotData

func (n *NatsProxy) SubscribeSnapshotData(sel *commonv1.EventSelector) (
	d <-chan *analysisv1.SnapshotData,
	q chan<- struct{},
	err error,
)

func (*NatsProxy) SubscribeSpeedmapData

func (n *NatsProxy) SubscribeSpeedmapData(sel *commonv1.EventSelector) (
	<-chan *racestatev1.PublishSpeedmapRequest,
	chan<- struct{},
	error,
)

type Option

type Option func(*NatsProxy)

func WithContext

func WithContext(ctx context.Context) Option

func WithLogger

func WithLogger(l *log.Logger) Option