sensor

Documentation

Index

• Constants
• Variables
• type DependencyDefinitionValue
• type JetstreamTriggerConn
  • func NewJetstreamTriggerConn(conn *jetstreambase.JetstreamConnection, sensorName string, triggerName string, ...) (*JetstreamTriggerConn, error)
  • func (conn *JetstreamTriggerConn) String() string
  • func (conn *JetstreamTriggerConn) Subscribe(ctx context.Context, closeCh <-chan struct{}, ...) error
• type MsgInfo
• type SensorJetstream
  • func NewSensorJetstream(url string, sensorSpec *v1alpha1.Sensor, streamConfig string, ...) (*SensorJetstream, error)
  • func (stream *SensorJetstream) Connect(triggerName string, dependencyExpression string, ...) (eventbuscommon.TriggerConnection, error)
  • func (stream *SensorJetstream) Initialize() error
• type TriggerValue

Constants

const (
	SensorNilError = "sensorSpec == nil??"
)

Variables

var (
	TriggersKey       = "Triggers"
	DependencyDefsKey = "Deps"
)

//////////////////////////////////////////////////////////////////////////////////////////////////// These are the Keys and methods to derive Keys for our K/V store

Types

type DependencyDefinitionValue

type DependencyDefinitionValue map[string]uint64 // value for DependencyDefsKey

//////////////////////////////////////////////////////////////////////////////////////////////////// These are the structs representing Values in our K/V store

type JetstreamTriggerConn

type JetstreamTriggerConn struct {
	*jetstreambase.JetstreamConnection
	// contains filtered or unexported fields
}

func NewJetstreamTriggerConn

func NewJetstreamTriggerConn(conn *jetstreambase.JetstreamConnection,
	sensorName string,
	triggerName string,
	dependencyExpression string,
	deps []eventbuscommon.Dependency) (*JetstreamTriggerConn, error)

func (*JetstreamTriggerConn) String

func (conn *JetstreamTriggerConn) String() string

func (*JetstreamTriggerConn) Subscribe

func (conn *JetstreamTriggerConn) Subscribe(ctx context.Context,
	closeCh <-chan struct{},
	resetConditionsCh <-chan struct{},
	lastResetTime time.Time,
	transform func(depName string, event cloudevents.Event) (*cloudevents.Event, error),
	filter func(string, cloudevents.Event) bool,
	action func(map[string]cloudevents.Event),
	defaultSubject *string) error

type MsgInfo

type MsgInfo struct {
	StreamSeq   uint64
	ConsumerSeq uint64
	Timestamp   time.Time
	Event       *cloudevents.Event
}

value for getDependencyKey()

type SensorJetstream

type SensorJetstream struct {
	*eventbusjetstreambase.Jetstream
	// contains filtered or unexported fields
}

func NewSensorJetstream

func NewSensorJetstream(url string, sensorSpec *v1alpha1.Sensor, streamConfig string, auth *eventbuscommon.Auth, logger *zap.SugaredLogger) (*SensorJetstream, error)

func (*SensorJetstream) Connect

func (stream *SensorJetstream) Connect(triggerName string, dependencyExpression string, deps []eventbuscommon.Dependency) (eventbuscommon.TriggerConnection, error)

func (*SensorJetstream) Initialize

func (stream *SensorJetstream) Initialize() error

type TriggerValue

type TriggerValue []string // value for TriggersKey