You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: producer.go
    path of file: ./repos/plumber/vendor/github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream
    
    the code of the file until where you have to start completion: package stream

import (
	"bufio"
	"bytes"
	"fmt"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/logs"
	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
	"sync"
	"sync/atomic"
	"time"
)

type ConfirmationStatus struct {
	message      message.StreamMessage
	producerID   uint8
	publishingId int64
	confirmed    bool
	err          error
	errorCode    uint16
	linkedTo     []*ConfirmationStatus
}

func (cs *ConfirmationStatus) IsConfirmed() bool {
	return cs.confirmed
}

func (cs *ConfirmationStatus) GetProducerID() uint8 {
	return cs.producerID
}

func (cs *ConfirmationStatus) GetPublishingId() int64 {
	return cs.publishingId
}

func (cs *ConfirmationStatus) GetError() error {
	return cs.err
}

func (cs *ConfirmationStatus) LinkedMessages() []*ConfirmationStatus {
	return cs.linkedTo
}

func (cs *ConfirmationStatus) GetMessage() message.StreamMessage {
	return cs.message
}

func (cs *ConfirmationStatus) GetErrorCode() uint16 {
	return cs.errorCode
}

type pendingMessagesSequence struct {
	messages []messageSequence
	size     int
}

type messageSequence st
    
func (producer *Producer) BatchSend(batchMessages []message.StreamMessage) error {
	var messagesSequence = make([]messageSequence, len(batchMessages))
	totalBufferToSend := 0
	for i, batchMessage := range batchMessages {
		messageBytes, err := batchMessage.MarshalBinary()
		if err != nil {
			return err
		}
		sequence := producer.assignPublishingID(batchMessage)
		totalBufferToSend += len(messageBytes)
		messagesSequence[i] = messageSequence{
			messageBytes:     messageBytes,
			unCompressedSize: len(messageBytes),
			publishingId:     sequence,
		}

		producer.addUnConfirmed(sequence, batchMessage, producer.id)
	}

	if totalBufferToSend+initBufferPublishSize > producer.options.client.tuneState.requestedMaxFrameSize {
		for _, msg := range messagesSequence {

			unConfirmedMessage := producer.getUnConfirmed(msg.publishingId)

			//producer.mutex.Lock()
			if producer.publishConfirm != nil {
				unConfirmedMessage.err = FrameTooLarge
				unConfirmedMessage.errorCode = responseCodeFrameTooLarge
				producer.publishConfirm <- []*ConfirmationStatus{unConfirmedMessage}
			}
			//producer.mutex.Unlock()
			producer.removeUnConfirmed(msg.publishingId)
		}

		return FrameTooLarge
	}

	return producer.internalBatchSend(messagesSequence)
}