pubsub

Documentation

Index

• func TopicWritingRule(option topic.Option, fragments []uint) func() []uint
• type Publisher
  • func NewPublisher(id string, address string, db *storage.DB, ...) Publisher
  • func (p *Publisher) StartTopicPublication(ctx context.Context, topicName string, retentionPeriodSec uint64, ...) (chan error, error)
  • func (p *Publisher) Subscribe(subscription *pb.Subscription, stream pb.PubSub_SubscribeServer) error
  • func (p *Publisher) Wait()
• type RetrievablePublisher
  • func NewRetrievablePublisher(id string, address string, db *storage.DB, ...) RetrievablePublisher
  • func (p *RetrievablePublisher) RetrievableSubscribe(stream pb.RetrievablePubSub_RetrievableSubscribeServer) error
  • func (p *RetrievablePublisher) StartTopicPublication(ctx context.Context, topicName string, retentionPeriodSec uint64, ...) (chan []TopicDataResult, chan error, error)
  • func (p *RetrievablePublisher) Wait()
• type RetrievableSubscriber
  • func NewRetrievableSubscriber(id string, bootstrapper *bootstrapping.BootstrapService, ...) RetrievableSubscriber
  • func (s *RetrievableSubscriber) StartTopicSubscription(ctx context.Context, topicName string, batchSize, flushInterval uint32) (chan RetrievableSubscriptionResults, chan error, error)
  • func (s *RetrievableSubscriber) Wait()
• type RetrievableSubscriptionResults
• type Subscriber
  • func NewSubscriber(id string, bootstrapper *bootstrapping.BootstrapService, ...) Subscriber
  • func (s *Subscriber) StartTopicSubscription(ctx context.Context, topicName string, batchSize, flushInterval uint32) (chan []SubscriptionResult, chan error, error)
  • func (s *Subscriber) Wait()
• type SubscriptionAddrs
• type SubscriptionResult
• type TopicData
• type TopicDataResult

Functions

func TopicWritingRule

func TopicWritingRule(option topic.Option, fragments []uint) func() []uint

Types

type Publisher

type Publisher struct {
	pb.PubSubServer
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(id string, address string, db *storage.DB, bootstrapper *bootstrapping.BootstrapService,
	publishedOffsets, fetchedOffsets storage.TopicFragmentOffsets) Publisher

func (*Publisher) StartTopicPublication

func (p *Publisher) StartTopicPublication(ctx context.Context, topicName string, retentionPeriodSec uint64,
	inStream chan TopicData) (chan error, error)

func (*Publisher) Subscribe

func (p *Publisher) Subscribe(subscription *pb.Subscription, stream pb.PubSub_SubscribeServer) error

func (*Publisher) Wait

func (p *Publisher) Wait()

type RetrievablePublisher

type RetrievablePublisher struct {
	pb.RetrievablePubSubServer
	// contains filtered or unexported fields
}

func NewRetrievablePublisher

func NewRetrievablePublisher(id string, address string, db *storage.DB, bootstrapper *bootstrapping.BootstrapService,
	publishedOffsets, fetchedOffsets storage.TopicFragmentOffsets) RetrievablePublisher

func (*RetrievablePublisher) RetrievableSubscribe

func (p *RetrievablePublisher) RetrievableSubscribe(stream pb.RetrievablePubSub_RetrievableSubscribeServer) error

func (*RetrievablePublisher) StartTopicPublication

func (p *RetrievablePublisher) StartTopicPublication(ctx context.Context, topicName string, retentionPeriodSec uint64,
	inStream chan TopicData) (chan []TopicDataResult, chan error, error)

func (*RetrievablePublisher) Wait

func (p *RetrievablePublisher) Wait()

type RetrievableSubscriber

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

func NewRetrievableSubscriber

func NewRetrievableSubscriber(id string, bootstrapper *bootstrapping.BootstrapService, subscribedOffsets storage.TopicFragmentOffsets) RetrievableSubscriber

func (*RetrievableSubscriber) StartTopicSubscription

func (s *RetrievableSubscriber) StartTopicSubscription(ctx context.Context, topicName string, batchSize, flushInterval uint32) (chan RetrievableSubscriptionResults, chan error, error)

func (*RetrievableSubscriber) Wait

func (s *RetrievableSubscriber) Wait()

type RetrievableSubscriptionResults

type RetrievableSubscriptionResults struct {
	Results  []SubscriptionResult
	SendBack func([]SubscriptionResult) error
}

type Subscriber

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

func NewSubscriber

func NewSubscriber(id string, bootstrapper *bootstrapping.BootstrapService, subscribedOffsets storage.TopicFragmentOffsets) Subscriber

func (*Subscriber) StartTopicSubscription

func (s *Subscriber) StartTopicSubscription(ctx context.Context, topicName string, batchSize, flushInterval uint32) (chan []SubscriptionResult, chan error, error)

func (*Subscriber) Wait

func (s *Subscriber) Wait()

type SubscriptionAddrs

type SubscriptionAddrs map[string][]uint

type SubscriptionResult

type SubscriptionResult struct {
	FragmentId uint
	SeqNum     uint64
	Data       []byte
}

type TopicData

type TopicData struct {
	SeqNum uint64
	Data   []byte
}

type TopicDataResult

type TopicDataResult struct {
	FragmentId uint32
	SeqNum     uint64
	Data       []byte
}