broker

Documentation

Index

• Variables
• type Broker
  • func New(id int32, opts ...BrokerFn) (*Broker, error)
  • func (b *Broker) AddPartition(partition *jocko.Partition) error
  • func (rm Broker) BecomeFollower(topic string, pid int32, command *protocol.PartitionState) error
  • func (rm Broker) BecomeLeader(topic string, pid int32, command *protocol.PartitionState) error
  • func (b *Broker) Cluster() []*jocko.ClusterMember
  • func (b *Broker) ClusterMember(id int32) *jocko.ClusterMember
  • func (b *Broker) ControllerID() string
  • func (b *Broker) CreateTopic(topic string, partitions int32) error
  • func (b *Broker) DeleteTopic(topic string) error
  • func (b *Broker) ID() int32
  • func (b *Broker) IsController() bool
  • func (b *Broker) IsLeaderOfPartition(topic string, pid int32, lid int32) bool
  • func (b *Broker) Join(addrs ...string) (int, error)
  • func (b *Broker) Partition(topic string, partition int32) (*jocko.Partition, error)
  • func (b *Broker) Shutdown() error
  • func (b *Broker) StartReplica(partition *jocko.Partition) error
  • func (b *Broker) TopicPartitions(topic string) (found []*jocko.Partition, err error)
  • func (b *Broker) Topics() []string
  • func (s *Broker) WaitForLeader(timeout time.Duration) (string, error)
• type BrokerFn
  • func Addr(brokerAddr string) BrokerFn
  • func LogDir(logDir string) BrokerFn
  • func Logger(logger *simplelog.Logger) BrokerFn
  • func Raft(raft jocko.Raft) BrokerFn
  • func Serf(serf jocko.Serf) BrokerFn
• type Replicator
  • func NewReplicator(partition *jocko.Partition, replicaID int32, opts ...ReplicatorFn) *Replicator
  • func (r *Replicator) Close() error
• type ReplicatorFn
  • func ReplicatorFetchSize(size int32) ReplicatorFn
  • func ReplicatorMaxWaitTime(time int32) ReplicatorFn
  • func ReplicatorMinBytes(size int32) ReplicatorFn
  • func ReplicatorProxy(proxy jocko.Proxy) ReplicatorFn
  • func ReplicatorReplicaID(id int32) ReplicatorFn

Variables

var (
	ErrTopicExists = errors.New("topic exists already")
)

Types

type Broker

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

func New

func New(id int32, opts ...BrokerFn) (*Broker, error)

func (*Broker) AddPartition

func (b *Broker) AddPartition(partition *jocko.Partition) error

func (Broker) BecomeFollower

func (rm Broker) BecomeFollower(topic string, pid int32, command *protocol.PartitionState) error

func (Broker) BecomeLeader

func (rm Broker) BecomeLeader(topic string, pid int32, command *protocol.PartitionState) error

func (*Broker) Cluster

func (b *Broker) Cluster() []*jocko.ClusterMember

func (*Broker) ClusterMember

func (b *Broker) ClusterMember(id int32) *jocko.ClusterMember

func (*Broker) ControllerID

func (b *Broker) ControllerID() string

func (*Broker) CreateTopic

func (b *Broker) CreateTopic(topic string, partitions int32) error

CreateTopic creates topic with partitions count.

func (*Broker) DeleteTopic

func (b *Broker) DeleteTopic(topic string) error

func (*Broker) ID

func (b *Broker) ID() int32

ID is used to get the broker's ID

func (*Broker) IsController

func (b *Broker) IsController() bool

IsController checks if this broker is the cluster controller

func (*Broker) IsLeaderOfPartition

func (b *Broker) IsLeaderOfPartition(topic string, pid int32, lid int32) bool

func (*Broker) Join

func (b *Broker) Join(addrs ...string) (int, error)

Join is used to have the broker join the gossip ring The target address should be another broker listening on the Serf address

func (*Broker) Partition

func (b *Broker) Partition(topic string, partition int32) (*jocko.Partition, error)

func (*Broker) Shutdown

func (b *Broker) Shutdown() error

func (*Broker) StartReplica

func (b *Broker) StartReplica(partition *jocko.Partition) error

func (*Broker) TopicPartitions

func (b *Broker) TopicPartitions(topic string) (found []*jocko.Partition, err error)

func (*Broker) Topics

func (b *Broker) Topics() []string

func (*Broker) WaitForLeader

func (s *Broker) WaitForLeader(timeout time.Duration) (string, error)

type BrokerFn

type BrokerFn func(b *Broker)

func Addr

func Addr(brokerAddr string) BrokerFn

func LogDir

func LogDir(logDir string) BrokerFn

func Logger

func Logger(logger *simplelog.Logger) BrokerFn

func Raft

func Raft(raft jocko.Raft) BrokerFn

func Serf

func Serf(serf jocko.Serf) BrokerFn

type Replicator

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

Replicator fetches from the partition's leader and produces to a follower thereby replicating the partition

func NewReplicator

func NewReplicator(partition *jocko.Partition, replicaID int32, opts ...ReplicatorFn) *Replicator

NewReplicator returns a new replicator object

func (*Replicator) Close

func (r *Replicator) Close() error

Close the replicator object when we are no longer following

type ReplicatorFn

type ReplicatorFn func(r *Replicator)

func ReplicatorFetchSize

func ReplicatorFetchSize(size int32) ReplicatorFn

func ReplicatorMaxWaitTime

func ReplicatorMaxWaitTime(time int32) ReplicatorFn

func ReplicatorMinBytes

func ReplicatorMinBytes(size int32) ReplicatorFn

func ReplicatorProxy

func ReplicatorProxy(proxy jocko.Proxy) ReplicatorFn

func ReplicatorReplicaID

func ReplicatorReplicaID(id int32) ReplicatorFn