broker

package

Go to main page

Versions in this module

v0

v0.1.7

Dec 10, 2017

Changes in this version

type Broker

+ func (b *Broker) NotAcknowledge(ctx context.Context, ...) (bool, error)

+ func (b *Broker) Produce(ctx context.Context, req *sgproto.ProduceMessageRequest) (*sgproto.ProduceResponse, error)

v0.1.6

Nov 19, 2017

Changes in this version

type Config

+ OffsetReplicationFactor int

v0.1.5

Nov 18, 2017

Changes in this version

+ var ErrPartitionNotFound = errors.New("ErrPartitionNotFound")

+ var RedeliveryTimeout = 10 * time.Second

type Broker

+ func (b *Broker) GetMarkStateMessage(ctx context.Context, ...) (*sgproto.Message, error)

+ func (b *Broker) MarkConsumed(ctx context.Context, ...) (bool, error)

v0.1.4

Oct 29, 2017

Changes in this version

type Config

+ LoggingLevel *logy.Level

v0.1.3

Oct 29, 2017

v0.1.2

Oct 29, 2017

v0.1.1

Oct 29, 2017

v0.1.0

Oct 28, 2017

Changes in this version

+ const ConsumerOffsetTopicName

+ var DefaultStateCheckInterval = 1 * time.Second

+ var ErrInvalidTopicName = errors.New("ErrInvalidTopicName")

+ var ErrNoConsumerFound = errors.New("ErrNoConsumerFound")

+ var ErrNoControllerSet = errors.New("ErrNoControllerSet")

+ var ErrNoKeySet = errors.New("ErrNoKeySet")

+ var ErrNoLeaderFound = errors.New("ErrNoLeaderFound")

+ var ErrNoPartitionSet = errors.New("ErrNoPartitionSet")

+ var ErrTopicAlreadyExist = errors.New("ErrTopicAlreadyExist")

+ var ErrTopicNotFound = errors.New("ErrTopicNotFound")

+ var ErrUnableToSelectReplicas = errors.New("ErrUnableToSelectReplicas")

+ type Broker struct

+ ShutdownCh chan struct{}

+ func New(conf *Config) (*Broker, error)

+ func (b *Broker) Acknowledge(ctx context.Context, ...) (bool, error)

+ func (b *Broker) AcknowledgeMessages(ctx context.Context, ...) error

+ func (b *Broker) Bootstrap() error

+ func (b *Broker) Commit(ctx context.Context, ...) (bool, error)

+ func (b *Broker) Conf() *Config

+ func (b *Broker) Consume(ctx context.Context, topicName, partition, consumerGroup, consumerName string, ...) error

+ func (b *Broker) CreateTopic(ctx context.Context, params *sgproto.CreateTopicParams) error

+ func (b *Broker) FetchFromSync(topicName, partition string, from []byte, fn func(msg *sgproto.Message) error) error

+ func (b *Broker) FetchRange(ctx context.Context, topicName, partition string, from, to sandflake.ID, ...) error

+ func (b *Broker) Get(ctx context.Context, topicName string, partition string, key []byte) (*sgproto.Message, error)

+ func (b *Broker) GetController() *sandglass.Node

+ func (b *Broker) GetTopic(name string) *topic.Topic

+ func (b *Broker) HasKey(ctx context.Context, topicName string, partition string, key []byte) (bool, error)

+ func (b *Broker) IsController() bool

+ func (b *Broker) Join(clusterAddrs ...string) error

+ func (b *Broker) LastOffset(ctx context.Context, ...) (sandflake.ID, error)

+ func (b *Broker) LaunchWatchers() error

+ func (b *Broker) Members() []*sandglass.Node

+ func (b *Broker) Name() string

+ func (b *Broker) PublishMessage(ctx context.Context, msg *sgproto.Message) (*sandflake.ID, error)

+ func (b *Broker) PublishMessages(ctx context.Context, msgs []*sgproto.Message) error

+ func (b *Broker) Stop(ctx context.Context) error

+ func (b *Broker) StoreMessageLocally(msg *sgproto.Message) error

+ func (b *Broker) StoreMessages(msgs []*sgproto.Message) error

+ func (b *Broker) Topics() []*topic.Topic

+ func (b *Broker) TriggerSyncRequest() error

+ func (b *Broker) WaitForIt() error

+ type Config struct

+ AdvertiseAddr string

+ BindAddr string

+ BootstrapRaft bool

+ DBPath string

+ DCName string

+ GRPCPort string

+ GossipPort string

+ HTTPPort string

+ InitialPeers []string

+ Name string

+ RaftPort string

+ type ConsumerGroup struct

+ func NewConsumerGroup(b *Broker, topic string, partition *topic.Partition, name string) *ConsumerGroup

+ func (c *ConsumerGroup) Consume(consumerName string) (<-chan *sgproto.Message, chan<- struct{}, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL