synapse

Documentation

Index

• Constants
• type Broker
  • func NewBroker(id int, namespace string, filepath string, writeBufferSize int) *Broker
  • func (b *Broker) AdvanceReadIndex() error
  • func (b *Broker) Initialize() error
  • func (b *Broker) PeekOne() ([]byte, error)
  • func (b *Broker) Print()
  • func (b *Broker) Write(data []byte) error
• type Command
• type CommandType
• type LagSnapshot
• type Server
  • func NewServer(port int, filepath string, bufferSize int, log *zap.Logger) *Server
  • func (s *Server) GetLagSnapshot() []LagSnapshot
  • func (s *Server) SetupRaft(serverID, advertiseAddr, bindAddr string, bootstrapPeers map[string]string) error
  • func (s *Server) Start() error

Constants

const (
	DATA_FILE_NAME               = "data"
	OFFSET_FILE_NAME             = "offset"
	INDEX_FILE_NAME              = "index"
	OFFSET_BUFFER_MAX_SIZE int64 = 4096
)

Types

type Broker

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

func NewBroker

func NewBroker(id int, namespace string, filepath string, writeBufferSize int) *Broker

func (*Broker) AdvanceReadIndex

func (b *Broker) AdvanceReadIndex() error

used by Raft to commit a read across all nodes

func (*Broker) Initialize

func (b *Broker) Initialize() error

func (*Broker) PeekOne

func (b *Broker) PeekOne() ([]byte, error)

func (*Broker) Print

func (b *Broker) Print()

func (*Broker) Write

func (b *Broker) Write(data []byte) error

type Command

type Command struct {
	Type      CommandType
	Namespace string
	Data      []byte
}

type CommandType

type CommandType byte

const (
	CmdProduce CommandType = 0x01
	CmdConsume CommandType = 0x02
)

type LagSnapshot

type LagSnapshot struct {
	Namespace string `json:"namespace"`
	Lag       int64  `json:"lag"`
}

type Server

type Server struct {
	Port    int
	Brokers map[string]*Broker
	// contains filtered or unexported fields
}

func NewServer

func NewServer(port int, filepath string, bufferSize int, log *zap.Logger) *Server

func (*Server) GetLagSnapshot

func (s *Server) GetLagSnapshot() []LagSnapshot

func (*Server) SetupRaft

func (s *Server) SetupRaft(serverID, advertiseAddr, bindAddr string, bootstrapPeers map[string]string) error

func (*Server) Start

func (s *Server) Start() error