broker

Documentation

Index

• Variables
• type Broker
  • func New(id int32, opts ...BrokerFn) (*Broker, error)
  • func (b *Broker) Join(addrs ...string) protocol.Error
  • func (b *Broker) Run(ctx context.Context, requestc <-chan jocko.Request, ...)
  • func (b *Broker) Shutdown() error
  • func (s *Broker) WaitForLeader(timeout time.Duration) (string, error)
• type BrokerFn
  • func Addr(brokerAddr string) BrokerFn
  • func LogDir(logDir string) BrokerFn
  • func Logger(logger jocko.Logger) BrokerFn
  • func Loner() BrokerFn
  • func Raft(raft jocko.Raft) BrokerFn
  • func RaftCommands(commands chan jocko.RaftCommand) BrokerFn
  • func Serf(serf jocko.Serf) BrokerFn
• type Replicator
  • func NewReplicator(partition *jocko.Partition, replicaID int32, opts ...ReplicatorFn) *Replicator
  • func (r *Replicator) Close() error
  • func (r *Replicator) Replicate()
• type ReplicatorFn
  • func ReplicatorFetchSize(size int32) ReplicatorFn
  • func ReplicatorLeader(leader jocko.Client) ReplicatorFn
  • func ReplicatorMaxWaitTime(time int32) ReplicatorFn
  • func ReplicatorMinBytes(size int32) ReplicatorFn
  • func ReplicatorReplicaID(id int32) ReplicatorFn

Variables

var (
	ErrTopicExists     = errors.New("topic exists already")
	ErrInvalidArgument = errors.New("no logger set")
)

Types

type Broker

type Broker struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Broker represents a broker in a Jocko cluster, like a broker in a Kafka cluster.

func New

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

New is used to instantiate a new broker.

func (*Broker) Join

func (b *Broker) Join(addrs ...string) protocol.Error

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

func (*Broker) Run

func (b *Broker) Run(ctx context.Context, requestc <-chan jocko.Request, responsec chan<- jocko.Response)

Run starts a loop to handle requests send back responses.

func (*Broker) Shutdown

func (b *Broker) Shutdown() error

Shutdown is used to shutdown the broker, its serf, its raft, and so on.

func (*Broker) WaitForLeader

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

type BrokerFn

type BrokerFn func(b *Broker)

BrokerFn is used to configure brokers.

func Addr

func Addr(brokerAddr string) BrokerFn

Addr is used to set the broker's client addr.

func LogDir

func LogDir(logDir string) BrokerFn

LogDir is used to set the directory the broker stores its data logs.

func Logger

func Logger(logger jocko.Logger) BrokerFn

Logger is used to set the broker's logger.

func Loner

func Loner() BrokerFn

Set broker as a loner for tests.

func Raft

func Raft(raft jocko.Raft) BrokerFn

Raft is used to set the broker's raft instance.

func RaftCommands

func RaftCommands(commands chan jocko.RaftCommand) BrokerFn

RaftCommands sets the channel the broker listens for raft commands, probably only useful for Jocko tests.

func Serf

func Serf(serf jocko.Serf) BrokerFn

Serf is used to set the broker's serf instance.

type Replicator

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

Replicator fetches from the partition's leader producing to itself the follower, thereby replicating the partition.

func NewReplicator

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

NewReplicator returns a new replicator instance.

func (*Replicator) Close

func (r *Replicator) Close() error

Close the replicator object when we are no longer following

func (*Replicator) Replicate

func (r *Replicator) Replicate()

type ReplicatorFn

type ReplicatorFn func(r *Replicator)

ReplicatorFn is used to configure replicators.

func ReplicatorFetchSize

func ReplicatorFetchSize(size int32) ReplicatorFn

ReplicatorFetchSize is used to set replicator's fetch request size.

func ReplicatorLeader

func ReplicatorLeader(leader jocko.Client) ReplicatorFn

ReplicatorLeader is used to set the replicator's leader to consume from.

func ReplicatorMaxWaitTime

func ReplicatorMaxWaitTime(time int32) ReplicatorFn

ReplicatorMaxWaitTime is used to set the replicator's request's max wait time. Similar to the consumer config in Kakfa.

func ReplicatorMinBytes

func ReplicatorMinBytes(size int32) ReplicatorFn

ReplicatorMinBytes is used to set the replicator's min byte request size. Similar to the consumer config in Kafka.

func ReplicatorReplicaID

func ReplicatorReplicaID(id int32) ReplicatorFn

ReplicatorReplicaID is used to set the ID of the broker this replicator should replicate. Similar to the consumer config in Kafka.