subscriber

Documentation

Index

• Constants
• type InitialLoadOptions
• type Message
  • func (msg *Message) Ack()
• type MessageHandler
• type Options
  • func NewOptions() *Options
• type Pipeline
  • func NewPipeline(subscriber *Subscriber, id uint64, lastSeq uint64) *Pipeline
  • func (pipeline *Pipeline) Pull() error
  • func (pipeline *Pipeline) UpdateLastSequence(sequence uint64)
• type PipelineState
• type Snapshot
  • func NewSnapshot(pipeline *Pipeline) *Snapshot
  • func (snapshot *Snapshot) Create() error
  • func (snapshot *Snapshot) Pull() error
• type StateStore
• type Subscriber
  • func NewSubscriber(options *Options) *Subscriber
  • func NewSubscriberWithClient(client *core.Client, options *Options) *Subscriber
  • func (sub *Subscriber) AddAllPipelines() error
  • func (sub *Subscriber) AddPipeline(pipeline *Pipeline) error
  • func (sub *Subscriber) Connect(host string, options *core.Options) error
  • func (sub *Subscriber) Disconnect()
  • func (sub *Subscriber) GetCollectionInfo(collection string) []string
  • func (sub *Subscriber) GetPipelineCount() (uint64, error)
  • func (sub *Subscriber) Register(subscriberType subscriber_manager_pb.SubscriberType, component string, ...) error
  • func (sub *Subscriber) Subscribe(cb MessageHandler) (*Subscription, error)
  • func (sub *Subscriber) SubscribeToCollections(colMap map[string][]string) error
• type Subscription
  • func NewSubscription(subscriber *Subscriber, bufferSize int) *Subscription
  • func (s *Subscription) Unsubscribe() error

Constants

const (
	SubscriberType_Transmitter subscriber_manager_pb.SubscriberType = subscriber_manager_pb.SubscriberType_TRANSMITTER
	SubscriberType_Exporter    subscriber_manager_pb.SubscriberType = subscriber_manager_pb.SubscriberType_EXPORTER
)

Types

type InitialLoadOptions

type InitialLoadOptions struct {
	Enabled      bool
	OmittedCount uint64
}

type Message

type Message struct {
	PipelineID   uint64
	Subscription *Subscription
	Event        *gravity_sdk_types_event.Event
}

func (*Message) Ack

func (msg *Message) Ack()

type MessageHandler

type MessageHandler func(*Message)

type Options

type Options struct {
	WorkerCount int
	BufferSize  int
	ChunkSize   int
	Verbose     bool
	StateStore  StateStore
	InitialLoad InitialLoadOptions
}

func NewOptions

func NewOptions() *Options

type Pipeline

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

func NewPipeline

func NewPipeline(subscriber *Subscriber, id uint64, lastSeq uint64) *Pipeline

func (*Pipeline) Pull

func (pipeline *Pipeline) Pull() error

func (*Pipeline) UpdateLastSequence

func (pipeline *Pipeline) UpdateLastSequence(sequence uint64)

type PipelineState

type PipelineState interface {
	GetLastSequence() uint64
	UpdateLastSequence(uint64) error
}

type Snapshot

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

func NewSnapshot

func NewSnapshot(pipeline *Pipeline) *Snapshot

func (*Snapshot) Create

func (snapshot *Snapshot) Create() error

func (*Snapshot) Pull

func (snapshot *Snapshot) Pull() error

type StateStore

type StateStore interface {
	GetPipelineState(uint64) (PipelineState, error)
	GetPipelines() []uint64
}

type Subscriber

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

func NewSubscriber

func NewSubscriber(options *Options) *Subscriber

func NewSubscriberWithClient

func NewSubscriberWithClient(client *core.Client, options *Options) *Subscriber

func (*Subscriber) AddAllPipelines

func (sub *Subscriber) AddAllPipelines() error

func (*Subscriber) AddPipeline

func (sub *Subscriber) AddPipeline(pipeline *Pipeline) error

func (*Subscriber) Connect

func (sub *Subscriber) Connect(host string, options *core.Options) error

func (*Subscriber) Disconnect

func (sub *Subscriber) Disconnect()

func (*Subscriber) GetCollectionInfo

func (sub *Subscriber) GetCollectionInfo(collection string) []string

func (*Subscriber) GetPipelineCount

func (sub *Subscriber) GetPipelineCount() (uint64, error)

func (*Subscriber) Register

func (sub *Subscriber) Register(subscriberType subscriber_manager_pb.SubscriberType, component string, subscriberID string, name string) error

func (*Subscriber) Subscribe

func (sub *Subscriber) Subscribe(cb MessageHandler) (*Subscription, error)

func (*Subscriber) SubscribeToCollections

func (sub *Subscriber) SubscribeToCollections(colMap map[string][]string) error

type Subscription

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

func NewSubscription

func NewSubscription(subscriber *Subscriber, bufferSize int) *Subscription

func (*Subscription) Unsubscribe

func (s *Subscription) Unsubscribe() error