broker

Documentation

Index

• Constants
• Variables
• type Broker
  • func New(id int32, opts ...BrokerFn) (*Broker, error)
  • func (s *Broker) AddBroker(broker jocko.BrokerConn) error
  • func (s *Broker) AddPartition(partition *jocko.Partition) error
  • func (s *Broker) Apply(l *raft.Log) interface{}
  • 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 (s *Broker) BrokerConn(id int32) *jocko.BrokerConn
  • func (b *Broker) Cluster() []*jocko.BrokerConn
  • func (s *Broker) ControllerID() string
  • func (s *Broker) CreateTopic(topic string, partitions int32) error
  • func (s *Broker) DeleteTopic(topic string) error
  • func (s *Broker) DeleteTopics(topics ...string) error
  • func (b *Broker) Host() string
  • func (b *Broker) ID() int32
  • func (s *Broker) IsController() bool
  • func (s *Broker) IsLeaderOfPartition(topic string, pid int32, lid int32) bool
  • func (b *Broker) IsShutdown() bool
  • func (s *Broker) Join(addrs ...string) (int, error)
  • func (b *Broker) Leave() error
  • func (s *Broker) Partition(topic string, partition int32) (*jocko.Partition, error)
  • func (b *Broker) Port() int
  • func (s *Broker) Restore(rc io.ReadCloser) error
  • func (b *Broker) Shutdown() error
  • func (s *Broker) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Broker) StartReplica(partition *jocko.Partition) error
  • func (s *Broker) TopicPartitions(topic string) (found []*jocko.Partition, err error)
  • func (s *Broker) Topics() []string
  • func (s *Broker) WaitForAppliedIndex(idx uint64, timeout time.Duration) error
  • func (s *Broker) WaitForLeader(timeout time.Duration) (string, error)
• type BrokerFn
  • func BindAddr(bindAddr string) BrokerFn
  • func Brokers(brokers []*jocko.BrokerConn) BrokerFn
  • func DataDir(dataDir string) BrokerFn
  • func LogDir(logDir string) BrokerFn
  • func Logger(logger *simplelog.Logger) BrokerFn
  • func Port(port int) BrokerFn
  • func RaftConfig(raft *raft.Config) BrokerFn
  • func RaftPort(raftPort int) BrokerFn
  • func SerfPort(serfPort int) BrokerFn
• type CmdType
• type FSMSnapshot
  • func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error
  • func (f *FSMSnapshot) Release()
• type ReplicatorFn
  • func ReplicatorFetchSize(size int32) ReplicatorFn
  • func ReplicatorMaxWaitTime(time int32) ReplicatorFn
  • func ReplicatorMinBytes(size int32) ReplicatorFn
  • func ReplicatorReplicaID(id int32) ReplicatorFn

Constants

const (
	// StatusReap is used to update the status of a node if we
	// are handling a EventMemberReap
	StatusReap = serf.MemberStatus(-1)
)

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) AddBroker

func (s *Broker) AddBroker(broker jocko.BrokerConn) error

func (*Broker) AddPartition

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

func (*Broker) Apply

func (s *Broker) Apply(l *raft.Log) interface{}

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) BrokerConn

func (s *Broker) BrokerConn(id int32) *jocko.BrokerConn

func (*Broker) Cluster

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

func (*Broker) ControllerID

func (s *Broker) ControllerID() string

func (*Broker) CreateTopic

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

CreateTopic creates topic with partitions count.

func (*Broker) DeleteTopic

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

func (*Broker) DeleteTopics

func (s *Broker) DeleteTopics(topics ...string) error

DeleteTopics creates topic with partitions count.

func (*Broker) Host

func (b *Broker) Host() string

Host is used to get Broker's host

func (*Broker) ID

func (b *Broker) ID() int32

ID is used to get the broker's ID

func (*Broker) IsController

func (s *Broker) IsController() bool

IsController checks if this broker is the cluster controller

func (*Broker) IsLeaderOfPartition

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

func (*Broker) IsShutdown

func (b *Broker) IsShutdown() bool

func (*Broker) Join

func (s *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) Leave

func (b *Broker) Leave() error

Leave is used to prepare for a graceful shutdown of the server

func (*Broker) Partition

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

func (*Broker) Port

func (b *Broker) Port() int

func (*Broker) Restore

func (s *Broker) Restore(rc io.ReadCloser) error

func (*Broker) Shutdown

func (b *Broker) Shutdown() error

func (*Broker) Snapshot

func (s *Broker) Snapshot() (raft.FSMSnapshot, error)

func (*Broker) StartReplica

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

func (*Broker) TopicPartitions

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

func (*Broker) Topics

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

func (*Broker) WaitForAppliedIndex

func (s *Broker) WaitForAppliedIndex(idx uint64, timeout time.Duration) error

func (*Broker) WaitForLeader

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

type BrokerFn

type BrokerFn func(b *Broker)

func BindAddr

func BindAddr(bindAddr string) BrokerFn

func Brokers

func Brokers(brokers []*jocko.BrokerConn) BrokerFn

func DataDir

func DataDir(dataDir string) BrokerFn

func LogDir

func LogDir(logDir string) BrokerFn

func Logger

func Logger(logger *simplelog.Logger) BrokerFn

func Port

func Port(port int) BrokerFn

func RaftConfig

func RaftConfig(raft *raft.Config) BrokerFn

func RaftPort

func RaftPort(raftPort int) BrokerFn

func SerfPort

func SerfPort(serfPort int) BrokerFn

type CmdType

type CmdType int

type FSMSnapshot

type FSMSnapshot struct {
}

func (*FSMSnapshot) Persist

func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error

func (*FSMSnapshot) Release

func (f *FSMSnapshot) Release()

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 ReplicatorReplicaID

func ReplicatorReplicaID(id int32) ReplicatorFn