mirror of
https://github.com/chrislusf/seaweedfs
synced 2025-07-26 13:32:46 +02:00
53 lines
1.1 KiB
Go
53 lines
1.1 KiB
Go
package messages
|
|
|
|
import (
|
|
flatbuffers "github.com/google/flatbuffers/go"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/segment"
|
|
)
|
|
|
|
type MessageBuffer struct {
|
|
fbsBuffer *flatbuffers.Builder
|
|
sequenceBase int64
|
|
counter int64
|
|
bb *segment.MessageBatchBuilder
|
|
isSealed bool
|
|
}
|
|
|
|
func NewMessageBuffer() *MessageBuffer {
|
|
t := &MessageBuffer{
|
|
fbsBuffer: flatbuffers.NewBuilder(4 * 1024 * 1024),
|
|
}
|
|
t.bb = segment.NewMessageBatchBuilder(t.fbsBuffer)
|
|
return t
|
|
}
|
|
|
|
func (mb *MessageBuffer) Reset(sequenceBase int64) {
|
|
mb.sequenceBase = sequenceBase
|
|
mb.counter = 0
|
|
mb.bb.Reset()
|
|
}
|
|
|
|
func (mb *MessageBuffer) AddMessage(message *Message) {
|
|
mb.bb.AddMessage(mb.sequenceBase, message.Ts.UnixMilli(), message.Properties, message.Key, message.Content)
|
|
mb.sequenceBase++
|
|
mb.counter++
|
|
}
|
|
|
|
func (mb *MessageBuffer) Len() int {
|
|
return int(mb.counter)
|
|
}
|
|
|
|
func (mb *MessageBuffer) Seal(producerId int32,
|
|
producerEpoch int32,
|
|
segmentId int32,
|
|
flags int32) {
|
|
mb.isSealed = true
|
|
mb.bb.BuildMessageBatch(producerId, producerEpoch, segmentId, flags)
|
|
}
|
|
|
|
func (mb *MessageBuffer) Bytes() []byte {
|
|
if !mb.isSealed {
|
|
return nil
|
|
}
|
|
return mb.bb.GetBytes()
|
|
}
|