Method writeStreamEntryContent

core/stream.go:465–556

writeStreamEntryContent writes a stream entry content as listpack

(entry *model.StreamEntry)

// writeStreamEntryContent writes a stream entry content as listpack
func (enc *Encoder) writeStreamEntryContent(entry *model.StreamEntry) error {
	// Calculate total messages (including deleted ones)
	totalMsgs := len(entry.Msgs)
	deletedCount := 0
	for _, msg := range entry.Msgs {
		if msg.Deleted {
			deletedCount++
		}
	}
	validCount := totalMsgs - deletedCount

	// Build listpack with proper backlen values
	var entries []listpackEntry

	// Add count and deleted count
	entries = append(entries, listpackEntry{intVal: int64(validCount)})
	entries = append(entries, listpackEntry{intVal: int64(deletedCount)})

	// Add master field names
	entries = append(entries, listpackEntry{intVal: int64(len(entry.Fields))})
	for _, field := range entry.Fields {
		entries = append(entries, listpackEntry{strVal: field})
	}
	// Add field count for master entry (this is what the decoder reads as "end flag")
	entries = append(entries, listpackEntry{strVal: strconv.Itoa(len(entry.Fields))})

	// Add messages
	for _, msg := range entry.Msgs {
		// Calculate flag
		flag := StreamItemFlagNone
		if msg.Deleted {
			flag |= StreamItemFlagDeleted
		}

		// Check if message uses same fields as master
		if len(msg.Fields) == len(entry.Fields) {
			sameFields := true
			for _, field := range entry.Fields {
				if _, exists := msg.Fields[field]; !exists {
					sameFields = false
					break
				}
			}
			if sameFields {
				flag |= StreamItemFlagSameFields
			}
		}

		// Add flag
		entries = append(entries, listpackEntry{intVal: int64(flag)})

		// Add message ID (relative to first message ID)
		msDiff := int64(msg.Id.Ms) - int64(entry.FirstMsgId.Ms)
		seqDiff := int64(msg.Id.Sequence) - int64(entry.FirstMsgId.Sequence)
		entries = append(entries, listpackEntry{intVal: msDiff})
		entries = append(entries, listpackEntry{intVal: seqDiff})

		// Add field count if not same fields