fqueue

type AssignTopicReq struct {
	Topic string `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	// 需要新建的分区 eg: [1, 3, 4]
	Partitions []uint32 `protobuf:"varint,2,rep,packed,name=partitions" json:"partitions,omitempty"`
	// 由接收此消息的broker做leader的partition, eg: [1, 4]
	LeaderPartitions []uint32 `protobuf:"varint,3,rep,packed,name=leaderPartitions" json:"leaderPartitions,omitempty"`
}

type Broker struct {
	Name string
	// rpc地址
	ListenerAddress string
	// etcd地址
	EtcdEndPoints []string

	// 存放数据的地址
	DataPath string
	// rpc
	RpcServer *grpc.Server
	//
	Topics map[string]*FileTopic
	// append chan, 收到别的broker发来的append
	AppendMsgChan chan *MsgBatch
	// error chan, 发生错误
	ErrorChan chan error
	// contains filtered or unexported fields
}

type BrokerConfig struct {
	Name string
	// rpc地址
	ListenerAddress string
	// etcd地址
	EtcdEndPoints []string
	// 存放数据的地址
	DataPath string

	Debug bool
}

type BrokerMember struct {
	Name            string
	ListenerAddress string
	Client          BrokerServiceClient
}

type BrokerServiceClient interface {
	// broker
	// append from other broker
	Append(ctx context.Context, opts ...grpc.CallOption) (BrokerService_AppendClient, error)
	// get msg
	Get(ctx context.Context, in *GetReq, opts ...grpc.CallOption) (*GetResp, error)
	// producer
	Push(ctx context.Context, opts ...grpc.CallOption) (BrokerService_PushClient, error)
	// 创建分区
	CreateTopic(ctx context.Context, in *CreateTopicReq, opts ...grpc.CallOption) (*Resp, error)
	// 分配分区
	AssignTopic(ctx context.Context, in *AssignTopicReq, opts ...grpc.CallOption) (*Resp, error)
	// consumer
	// subscribe topics
	Subscribe(ctx context.Context, in *SubReq, opts ...grpc.CallOption) (*SubResp, error)
	// Producer pull msg
	Pull(ctx context.Context, in *PullReq, opts ...grpc.CallOption) (BrokerService_PullClient, error)
}

type BrokerServiceServer interface {
	// broker
	// append from other broker
	Append(BrokerService_AppendServer) error
	// get msg
	Get(context.Context, *GetReq) (*GetResp, error)
	// producer
	Push(BrokerService_PushServer) error
	// 创建分区
	CreateTopic(context.Context, *CreateTopicReq) (*Resp, error)
	// 分配分区
	AssignTopic(context.Context, *AssignTopicReq) (*Resp, error)
	// consumer
	// subscribe topics
	Subscribe(context.Context, *SubReq) (*SubResp, error)
	// Producer pull msg
	Pull(*PullReq, BrokerService_PullServer) error
}

type BrokerService_AppendClient interface {
	Send(*MsgBatch) error
	CloseAndRecv() (*Resp, error)
	grpc.ClientStream
}

type BrokerService_AppendServer interface {
	SendAndClose(*Resp) error
	Recv() (*MsgBatch, error)
	grpc.ServerStream
}

type BrokerService_PullClient interface {
	Recv() (*MsgBatch, error)
	grpc.ClientStream
}

type BrokerService_PullServer interface {
	Send(*MsgBatch) error
	grpc.ServerStream
}

type BrokerService_PushClient interface {
	Send(*MsgBatch) error
	CloseAndRecv() (*Resp, error)
	grpc.ClientStream
}

type BrokerService_PushServer interface {
	SendAndClose(*Resp) error
	Recv() (*MsgBatch, error)
	grpc.ServerStream
}

type CreateTopicReq struct {
	Topic          string `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	PartitionCount uint32 `protobuf:"varint,2,opt,name=partitionCount" json:"partitionCount,omitempty"`
	ReplicaCount   uint32 `protobuf:"varint,3,opt,name=replicaCount" json:"replicaCount,omitempty"`
}

type EtcdBroker struct {
	Version uint32 `json:"version"`
	Address string `json:"address"`
}

type EtcdConsumer struct {
	Version      uint32              `json:"version"`
	Subscription map[string][]uint32 `json:"subscription"`
}

type EtcdConsumerGroup struct {
	Version uint32
	Topics  []string
}

type EtcdTopic struct {
	Version    uint32              `json:"version"`
	Partitions map[uint32][]string `json:"partitions"`
}

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

type FilePartition struct {
	Id uint32

	Index *FileIndex

	// 最新消息的offset
	LatestMsgOffset uint64

	// 刷盘
	FlushTime time.Duration
	// contains filtered or unexported fields
}

type FileTopic struct {
	Name            string
	PartitionIds    []uint32
	PartitionNumber uint32
	Partitions      map[uint32]*FilePartition
	// 存放，可能没用
	//MsgChan         []chan *Msg
	// 一次读取的数量
	BatchCount uint32
}

type GetReq struct {
	Topic       string `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	Partition   uint32 `protobuf:"varint,2,opt,name=partition" json:"partition,omitempty"`
	StartOffset uint64 `protobuf:"varint,3,opt,name=startOffset" json:"startOffset,omitempty"`
}

type GetResp struct {
	Resp *Resp     `protobuf:"bytes,1,opt,name=resp" json:"resp,omitempty"`
	Msgs *MsgBatch `protobuf:"bytes,2,opt,name=msgs" json:"msgs,omitempty"`
	// 如果跟上, 则不用继续发get
	Enough bool `protobuf:"varint,3,opt,name=enough" json:"enough,omitempty"`
}

type Index interface {
	// 获取消息位置
	Position(offset uint64) (position uint64, length uint32, err error)

	// 写入消息时包含长度
	WriteIndex(offset, position uint64, length uint32) error
}

type Msg struct {
	Offset  uint64 // 8 byte
	Size    uint32 // 4 byte
	Crc     uint32 // 4 byte
	Type    byte   // 1 byte
	Version byte   // 1 byte
	KeyLen  uint32 // 4 byte  // fixed_header_length = 22
	Key     []byte // KeyLen byte
	Value   []byte // Size = fixed_header_length + keyLen + ValueLen
	Source  []byte
}

type MsgBatch struct {
	Topic       string   `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	Partition   uint32   `protobuf:"varint,2,opt,name=partition" json:"partition,omitempty"`
	StartOffset uint64   `protobuf:"varint,3,opt,name=startOffset" json:"startOffset,omitempty"`
	Msgs        [][]byte `protobuf:"bytes,4,rep,name=msgs,proto3" json:"msgs,omitempty"`
}

type MsgIndex struct {
	Offset   uint64
	Position uint64
	Len      uint32
}

type Partition interface {
	// 写入新数据
	WriteMsg(*Msg) error

	WriteMultiMsg([]*Msg) error

	// 读取指定offset的数据
	ReadMsg(uint64) ([]byte, error)

	ReadMultiMsg(uint64, uint32) ([][]byte, error)

	OffsetLag(uint64) uint64
}

type PartitionConfig struct {
	Id       uint32
	Topic    string
	DataPath string
	// ms, flush write content to disk
	FlushTime uint32
}

type PartitionInfo struct {
	Partition uint32
	Leader    string
	Brokers   []string

	Topic  string
	Offset uint64
	// contains filtered or unexported fields
}

type PullReq struct {
	Count uint32 `protobuf:"varint,1,opt,name=count" json:"count,omitempty"`
	// 超时时间, ms, <=0代表立即返回
	Timeout int64 `protobuf:"varint,2,opt,name=timeout" json:"timeout,omitempty"`
	// 这些信息由客户端存储
	TpSet *TopicPartitionOffset `protobuf:"bytes,3,opt,name=tpSet" json:"tpSet,omitempty"`
}

type Resp struct {
	Status  RespStatus `protobuf:"varint,1,opt,name=status,enum=fqueue.RespStatus" json:"status,omitempty"`
	Comment string     `protobuf:"bytes,2,opt,name=comment" json:"comment,omitempty"`
}

type RespStatus int32

type SubReq struct {
	Topics []*TopicPartition `protobuf:"bytes,1,rep,name=topics" json:"topics,omitempty"`
}

type SubResp struct {
	Resp *Resp `protobuf:"bytes,1,opt,name=resp" json:"resp,omitempty"`
	// 分配给它的topic, partition, 以及最新的offset
	TopicPartitionOffset []*TopicPartitionOffset `protobuf:"bytes,2,rep,name=topicPartitionOffset" json:"topicPartitionOffset,omitempty"`
}

type Topic interface {
	// 写到指定的分区
	Write(msg *Msg, partition uint32) error
	// 批量写入同一分区，保证同时写入成功或者同时失败
	WriteMulti(msgs []*Msg, partition uint32) error
	// 读取单条消息
	Read(offset uint64, partition uint32) ([]byte, error)
	// 批量读取
	ReadMulti(offset uint64, partition, count uint32) ([][]byte, int)

	OffsetLag(offset uint64, partition uint32) uint64

	Close() error
}

type TopicConfig struct {
	Name         string
	PartitionIds []uint32
	BatchCount   uint32
	BasePath     string
}

type TopicPartition struct {
	Topic     string   `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	Partition []uint32 `protobuf:"varint,2,rep,packed,name=partition" json:"partition,omitempty"`
}

type TopicPartitionLeader struct {
	Topic           string            `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	PartitionLeader map[uint32]string `` /* 151-byte string literal not displayed */
}

type TopicPartitionOffset struct {
	Topic string `protobuf:"bytes,1,opt,name=topic" json:"topic,omitempty"`
	// partition, offset
	PartitionOffset map[uint32]uint64 `` /* 152-byte string literal not displayed */
}