broker

Documentation

Index

• Constants
• type Broker
  • func NewBroker(opts BrokerOpts) (*Broker, error)
  • func (b *Broker) AddPeerToList(peerDataArray []PeerData) error
  • func (b *Broker) AppendEntriesLoop(msg api.Message)
  • func (b *Broker) AppendEntriesRPC(args *api.AppendEntriesArgs, reply *api.AppendEntriesReply) error
  • func (b *Broker) AppendTopicLoop()
  • func (b *Broker) AppendTopicRPC(args *api.AppendTopicArgs, reply *api.AppendTopicReply) error
  • func (b *Broker) CheckIfPeerAlreadyInList(peer []PeerData) bool
  • func (b *Broker) ClearCluster() error
  • func (b *Broker) Consume(fetchReq api.FetchMessage) (api.Message, error)
  • func (b *Broker) CreateTopic(name string, replicationFactor int, partitionCount int) error
  • func (b *Broker) GetPeerList() ([]PeerData, error)
  • func (b *Broker) GetPeers() []Peer
  • func (b *Broker) InitPeers()
  • func (b *Broker) PeerRefreshLoop()
  • func (b *Broker) PrintRPC(args *int, reply *int) error
  • func (b *Broker) Produce(msg api.Message) error
  • func (b *Broker) Start()
  • func (b *Broker) StartRPCServer(wg *sync.WaitGroup)
  • func (b *Broker) StartServer()
• type BrokerOpts
• type CreateTopicRequest
• type Peer
• type PeerData

Constants

const (
	LEADER    = "leader"
	FOLLOWER  = "follower"
	CANDIDATE = "candidate"
)

Types

type Broker

type Broker struct {
	Name   string
	Peers  []Peer
	Topics []*topic.Topic

	Id          int
	Role        string
	CurrentTerm int
	VotedFor    int
	// contains filtered or unexported fields
}

func NewBroker

func NewBroker(opts BrokerOpts) (*Broker, error)

func (*Broker) AddPeerToList

func (b *Broker) AddPeerToList(peerDataArray []PeerData) error

Adds peer to list

func (*Broker) AppendEntriesLoop

func (b *Broker) AppendEntriesLoop(msg api.Message)

func (*Broker) AppendEntriesRPC

func (b *Broker) AppendEntriesRPC(args *api.AppendEntriesArgs, reply *api.AppendEntriesReply) error

func (*Broker) AppendTopicLoop

func (b *Broker) AppendTopicLoop()

func (*Broker) AppendTopicRPC

func (b *Broker) AppendTopicRPC(args *api.AppendTopicArgs, reply *api.AppendTopicReply) error

func (*Broker) CheckIfPeerAlreadyInList

func (b *Broker) CheckIfPeerAlreadyInList(peer []PeerData) bool

Checks if peer is already in maintained list

func (*Broker) ClearCluster

func (b *Broker) ClearCluster() error

Clears the cluster from etcd

func (*Broker) Consume

func (b *Broker) Consume(fetchReq api.FetchMessage) (api.Message, error)

Consunmer Manger functions

func (*Broker) CreateTopic

func (b *Broker) CreateTopic(name string, replicationFactor int, partitionCount int) error

CreateTopic creates a new topic

func (*Broker) GetPeerList

func (b *Broker) GetPeerList() ([]PeerData, error)

Equivalent of GetPeers function

func (*Broker) GetPeers

func (b *Broker) GetPeers() []Peer

func (*Broker) InitPeers

func (b *Broker) InitPeers()

func (*Broker) PeerRefreshLoop

func (b *Broker) PeerRefreshLoop()

Keeps checking if new peers have been added in the kv store

func (*Broker) PrintRPC

func (b *Broker) PrintRPC(args *int, reply *int) error

func (*Broker) Produce

func (b *Broker) Produce(msg api.Message) error

Producer Manager functions

func (*Broker) Start

func (b *Broker) Start()

func (*Broker) StartRPCServer

func (b *Broker) StartRPCServer(wg *sync.WaitGroup)

Starts the server for raft related RPC calls

func (*Broker) StartServer

func (b *Broker) StartServer()

type BrokerOpts

type BrokerOpts struct {
	Id        int    `json:"id"`
	Name      string `json:"name"`
	BrokerUri string `json:"broker_uri"`
	RpcUri    string `json:"rpc_uri"`
	Role      string `json:"role"`
}

type CreateTopicRequest

type CreateTopicRequest struct {
	Name              string `json:"name"`
	ReplicationFactor int    `json:"replicationFactor"`
	PartitionCount    int    `json:"partitionCount"`
}

type Peer

type Peer struct {
	Data PeerData
	// contains filtered or unexported fields
}

type PeerData

type PeerData struct {
	Id  int    `json:"id"`
	Uri string `json:"uri"`
}