internal

type BatchBuilder interface {
	// IsFull check if the size in the current batch exceeds the maximum size allowed by the batch
	IsFull() bool

	// Add will add single message to batch.
	Add(
		metadata *pb.SingleMessageMetadata, sequenceIDGenerator *uint64,
		payload []byte,
		callback interface{}, replicateTo []string, deliverAt time.Time,
		schemaVersion []byte, multiSchemaEnabled bool,
		useTxn bool,
		mostSigBits uint64,
		leastSigBits uint64,
	) bool

	// Flush all the messages buffered in the client and wait until all messages have been successfully persisted.
	Flush() *FlushBatch

	// FlushBatches all the messages buffered in multiple batches and wait until all
	// messages have been successfully persisted.
	FlushBatches() []*FlushBatch

	// Return the batch container batch message in multiple batches.
	IsMultiBatches() bool

	Close() error
	// contains filtered or unexported methods
}

type BatcherBuilderProvider func(
	maxMessages uint, maxBatchSize uint, maxMessageSize uint32, producerName string, producerID uint64,
	compressionType pb.CompressionType, level compression.Level,
	bufferPool BuffersPool, metrics *Metrics, logger log.Logger, encryptor crypto.Encryptor,
) (BatchBuilder, error)

type BlockingQueue interface {
	// Put enqueue one item, block if the queue is full
	// This is currently used for the internal testing
	Put(item interface{})

	// PutUnsafe enqueue one item without locking the queue, block if the queue is full
	PutUnsafe(item interface{})

	// Take dequeue one item, block until it's available
	Take() interface{}

	// Poll dequeue one item, return nil if queue is empty
	Poll() interface{}

	// CompareAndPoll compare the first item and poll it if meet the conditions
	CompareAndPoll(compare func(item interface{}) bool) interface{}

	// Peek return the first item without dequeing, return nil if queue is empty
	Peek() interface{}

	// PeekLast return last item in queue without dequeing, return nil if queue is empty
	PeekLast() interface{}

	// Size return the current size of the queue
	Size() int

	// ReadableSlice returns a new view of the readable items in the queue
	ReadableSlice() []interface{}

	// IterateUnsafe iterates the items in the queue without blocking the queue
	IterateUnsafe(func(item interface{}))

	// Lock locks the queue for manual control
	// Users must call Unlock() after finishing their operations
	Lock()

	// Unlock unlocks the queue
	// Must be called after Lock() to release the lock
	Unlock()
}

type Buffer interface {
	ReadableBytes() uint32

	WritableBytes() uint32

	// Capacity returns the capacity of the buffer's underlying byte slice,
	// that is, the total space allocated for the buffer's data.
	Capacity() uint32

	IsWritable() bool

	Read(size uint32) []byte

	Skip(size uint32)

	Get(readerIndex uint32, size uint32) []byte

	ReadableSlice() []byte

	WritableSlice() []byte

	// WrittenBytes advance the writer index when data was written in a slice
	WrittenBytes(size uint32)

	// MoveToFront copy the available portion of data at the beginning of the buffer
	MoveToFront()

	ReadUint16() uint16
	ReadUint32() uint32

	WriteUint16(n uint16)
	WriteUint32(n uint32)

	WriterIndex() uint32
	ReaderIndex() uint32

	Write(s []byte)

	Put(writerIdx uint32, s []byte)
	PutUint32(n uint32, writerIdx uint32)

	Resize(newSize uint32)
	ResizeIfNeeded(spaceNeeded uint32)

	// Retain increases the reference count
	Retain()
	// Release decreases the reference count and returns the buffer to the pool
	// if it's associated with a buffer pool and the count reaches zero.
	Release()
	// RefCnt returns the current reference count of the buffer.
	RefCnt() int64
	// SetReleaseCallback sets a callback function that will be called when the buffer is returned to a pool.
	SetReleaseCallback(cb func())

	// Clear will clear the current buffer data.
	Clear()
}

type BuffersPool interface {
	GetBuffer(initSize int) Buffer
	Put(buf Buffer)
}

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

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

type Closable interface {
	Close()
}

type Connection interface {
	SendRequest(requestID uint64, req *pb.BaseCommand, callback func(*pb.BaseCommand, error))
	SendRequestNoWait(req *pb.BaseCommand) error
	WriteData(ctx context.Context, data Buffer)
	RegisterListener(id uint64, listener ConnectionListener) error
	UnregisterListener(id uint64)
	AddConsumeHandler(id uint64, handler ConsumerHandler) error
	DeleteConsumeHandler(id uint64)
	ID() string
	GetMaxMessageSize() int32
	Close()
	WaitForClose() <-chan struct{}
	IsProxied() bool
}

type ConnectionListener interface {
	// ReceivedSendReceipt receive and process the return value of the send command.
	ReceivedSendReceipt(response *pb.CommandSendReceipt)

	// ConnectionClosed close the TCP connection.
	ConnectionClosed(closeProducer *pb.CommandCloseProducer)

	// SetRedirectedClusterURI set the redirected cluster URI for lookups
	SetRedirectedClusterURI(redirectedClusterURI string)
}

type ConnectionPool interface {
	// GetConnection get a connection from ConnectionPool.
	GetConnection(logicalAddr *url.URL, physicalAddr *url.URL, keySuffix int32) (Connection, error)

	// GetConnections get all connections in the pool.
	GetConnections() map[string]Connection

	// GenerateRoundRobinIndex generates a round-robin index.
	GenerateRoundRobinIndex() int32

	// Close all the connections in the pool
	Close()
}

type ConsumerHandler interface {
	MessageReceived(response *pb.CommandMessage, headersAndPayload Buffer) error

	ActiveConsumerChanged(isActive bool)

	// ConnectionClosed close the TCP connection.
	ConnectionClosed(closeConsumer *pb.CommandCloseConsumer)

	// SetRedirectedClusterURI set the redirected cluster URI for lookups
	SetRedirectedClusterURI(redirectedClusterURI string)
}

type FlushBatch struct {
	BatchData  Buffer
	SequenceID uint64
	Callbacks  []interface{}
	Error      error
}

type GetTopicsOfNamespaceMode string

type HTTPClient interface {
	Get(endpoint string, obj interface{}, params map[string]string) error
	Closable
}

type LeveledMetrics struct {
	MessagesPublished        prometheus.Counter
	BytesPublished           prometheus.Counter
	MessagesPending          prometheus.Gauge
	BytesPending             prometheus.Gauge
	PublishErrorsTimeout     prometheus.Counter
	PublishErrorsMsgTooLarge prometheus.Counter
	PublishLatency           prometheus.Observer
	PublishRPCLatency        prometheus.Observer

	MessagesReceived   prometheus.Counter
	BytesReceived      prometheus.Counter
	PrefetchedMessages prometheus.Gauge
	PrefetchedBytes    prometheus.Gauge
	AcksCounter        prometheus.Counter
	NacksCounter       prometheus.Counter
	DlqCounter         prometheus.Counter
	ProcessingTime     prometheus.Observer

	ProducersOpened            prometheus.Counter
	ProducersClosed            prometheus.Counter
	ProducersReconnectFailure  prometheus.Counter
	ProducersReconnectMaxRetry prometheus.Counter
	ProducersPartitions        prometheus.Gauge
	ConsumersOpened            prometheus.Counter
	ConsumersClosed            prometheus.Counter
	ConsumersReconnectFailure  prometheus.Counter
	ConsumersReconnectMaxRetry prometheus.Counter
	ConsumersPartitions        prometheus.Gauge
	ReadersOpened              prometheus.Counter
	ReadersClosed              prometheus.Counter
}

type LookupResult struct {
	LogicalAddr  *url.URL
	PhysicalAddr *url.URL
}

type LookupSchema struct {
	SchemaType SchemaType
	Data       []byte
	Properties map[string]string
}

type LookupService interface {
	// Lookup perform a lookup for the given topic, confirm the location of the broker
	// where the topic is located, and return the LookupResult.
	Lookup(topic string) (*LookupResult, error)

	// GetPartitionedTopicMetadata perform a CommandPartitionedTopicMetadata request for
	// the given topic, returns the CommandPartitionedTopicMetadataResponse as the result.
	GetPartitionedTopicMetadata(topic string) (*PartitionedTopicMetadata, error)

	// GetTopicsOfNamespace returns all the topics name for a given namespace.
	GetTopicsOfNamespace(namespace string, mode GetTopicsOfNamespaceMode) ([]string, error)

	// GetSchema returns schema for a given version.
	GetSchema(topic string, schemaVersion []byte) (*LookupSchema, error)

	GetBrokerAddress(brokerServiceURL string, proxyThroughServiceURL bool) (*LookupResult, error)

	ServiceNameResolver() *ServiceNameResolver

	// Closable Allow Lookup Service's internal client to be able to closed
	Closable
}

type MemoryLimitController interface {
	ReserveMemory(ctx context.Context, size int64) bool
	TryReserveMemory(size int64) bool
	ForceReserveMemory(size int64)
	ReleaseMemory(size int64)
	CurrentUsage() int64
	CurrentUsagePercent() float64
	IsMemoryLimited() bool
	RegisterTrigger(trigger func())
}

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

type Metrics struct {
	SendingBuffersCount prometheus.Gauge

	// Metrics that are not labeled with specificity are immediately available
	ConnectionsOpened                     prometheus.Counter
	ConnectionsClosed                     prometheus.Counter
	ConnectionsEstablishmentErrors        prometheus.Counter
	ConnectionsHandshakeErrors            prometheus.Counter
	LookupRequestsCount                   prometheus.Counter
	PartitionedTopicMetadataRequestsCount prometheus.Counter
	RPCRequestCount                       prometheus.Counter
	// contains filtered or unexported fields
}

type OpFn[T any] func() (T, error)

type PartitionedTopicMetadata struct {
	Partitions int `json:"partitions"` // Number of partitions for the topic
}

type PulsarServiceURI struct {
	ServiceName  string
	ServiceInfos []string
	ServiceHosts []string

	URL *url.URL
	// contains filtered or unexported fields
}

type RPCClient interface {
	// Create a new unique request id
	NewRequestID() uint64

	NewProducerID() uint64

	NewConsumerID() uint64

	// Send a request and block until the result is available
	RequestToAnyBroker(requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

	RequestToHost(serviceNameResolver *ServiceNameResolver, requestID uint64,
		cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

	RequestWithCnxKeySuffix(logicalAddr *url.URL, physicalAddr *url.URL, cnxKeySuffix int32, requestID uint64,
		cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

	Request(logicalAddr *url.URL, physicalAddr *url.URL, requestID uint64,
		cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

	RequestOnCnxNoWait(cnx Connection, cmdType pb.BaseCommand_Type, message proto.Message) error

	RequestOnCnx(cnx Connection, requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)

	LookupService(URL string) (LookupService, error)
}

type RPCResult struct {
	Response *pb.BaseCommand
	Cnx      Connection
}

type SchemaType int

type Semaphore interface {
	// Acquire a permit, if one is available and returns immediately,
	// reducing the number of available permits by one.
	Acquire(ctx context.Context) bool

	// Try to acquire a permit. The method will return immediately
	// with a `true` if it was possible to acquire a permit and
	// `false` otherwise.
	TryAcquire() bool

	// Release a permit, returning it to the semaphore.
	// Release a permit, increasing the number of available permits by
	// one.  If any threads are trying to acquire a permit, then one is
	// selected and given the permit that was just released.  That thread
	// is (re)enabled for thread scheduling purposes.
	// There is no requirement that a thread that releases a permit must
	// have acquired that permit by calling Acquire().
	// Correct usage of a semaphore is established by programming convention
	// in the application.
	Release()
}

type ServiceNameResolver interface {
	ResolveHost() (*url.URL, error)
	ResolveHostURI() (*PulsarServiceURI, error)
	UpdateServiceURL(url *url.URL) error
	GetServiceURI() *PulsarServiceURI
	GetServiceURL() *url.URL
	GetAddressList() []*url.URL
}

type TLSOptions struct {
	KeyFile                 string
	CertFile                string
	TrustCertsFilePath      string
	AllowInsecureConnection bool
	ValidateHostname        bool
	ServerName              string
	CipherSuites            []uint16
	MinVersion              uint16
	MaxVersion              uint16
	TLSConfig               *tls.Config
}

type TopicName struct {
	Domain    string
	Tenant    string
	Namespace string
	Topic     string
	Name      string
	Partition int
}