messagebroker

type AddTopicPartitionRequest struct {
	Name          string
	NumPartitions int
}

type AddTopicPartitionResponse struct {
	Response interface{}
}

type Admin interface {
	// CreateTopic creates a new topic if not available
	CreateTopic(context.Context, CreateTopicRequest) (CreateTopicResponse, error)

	// DeleteTopic deletes an existing topic
	DeleteTopic(context.Context, DeleteTopicRequest) (DeleteTopicResponse, error)

	// AddTopicPartitions adds partitions to an existing topic
	AddTopicPartitions(context.Context, AddTopicPartitionRequest) (*AddTopicPartitionResponse, error)

	// AlterTopicConfigs alters topic configuration
	AlterTopicConfigs(context.Context, ModifyTopicConfigRequest) ([]string, error)

	// DescribeTopicConfigs describes topic configuration
	DescribeTopicConfigs(context.Context, []string) (map[string]map[string]string, error)

	// IsHealthy checks health of underlying broker
	IsHealthy(context.Context) (bool, error)

	// FetchProjectTopics fetches a list of all topics for a given project
	FetchProjectTopics(ctx context.Context, project string) (map[string]bool, error)
}

type AdminClientOptions struct{}

type AdminConfig struct {
	EnableTopicCleanUp bool
}

type Broker interface {
	Admin
	Producer
	Consumer
}

type BrokerConfig struct {
	// A list of host/port pairs to use for establishing the initial connection to the broker cluster
	Brokers             []string
	EnableTLS           bool
	UserCertificate     string
	UserKey             string
	CACertificate       string
	CertDir             string
	Version             string
	DebugEnabled        bool
	OperationTimeoutMs  int
	ConnectionTimeoutMs int
	Producer            *ProducerConfig
	Consumer            *ConsumerConfig
	Admin               *AdminConfig
}

type CommitOnTopicRequest struct {
	Topic     string
	Partition int32
	Offset    int32
	ID        string
}

type CommitOnTopicResponse struct {
	Response interface{}
}

type Consumer interface {
	// ReceiveMessages gets tries to get the number of messages mentioned in the param "numOfMessages"
	// from the previous committed offset. If the available messages in the queue are less, returns
	// how many ever messages are available
	ReceiveMessages(context.Context, GetMessagesFromTopicRequest) (*GetMessagesFromTopicResponse, error)

	// CommitByPartitionAndOffset Commits messages if any
	// This func will commit the message consumed
	// by all the previous calls to GetMessages
	CommitByPartitionAndOffset(context.Context, CommitOnTopicRequest) (CommitOnTopicResponse, error)

	// CommitByMsgID Commits a message by ID
	CommitByMsgID(context.Context, CommitOnTopicRequest) (CommitOnTopicResponse, error)

	// GetTopicMetadata gets the topic metadata
	GetTopicMetadata(context.Context, GetTopicMetadataRequest) (GetTopicMetadataResponse, error)

	// Pause pause the consumer
	Pause(context.Context, PauseOnTopicRequest) error

	// Resume resume the consumer
	Resume(context.Context, ResumeOnTopicRequest) error

	// FetchConsumerLag returns the watermark and current offset for a consumer
	FetchConsumerLag(context.Context) (map[string]uint64, error)

	// Close closes the consumer
	Close(context.Context) error
}

type ConsumerClientOptions struct {
	// Specify a list of topics to consume messages from
	Topics []string
	// Specify the subscription name for this consumer. Only used for pulsar
	Subscription string
	// A unique string that identifies the consumer group this consumer belongs to.
	GroupID string
	// A unique identifier of the consumer instance provided by the end user.
	// Only non-empty strings are permitted. If set, the consumer is treated as a static member,
	// which means that only one instance with this ID is allowed in the consumer group at any time
	GroupInstanceID string
	// Specify the auto offset reset value in case no offset is committed yet.
	AutoOffsetReset string
}

type ConsumerConfig struct {
	SubscriptionType int
	OffsetReset      string
	EnableAutoCommit bool
}

type CreateTopicRequest struct {
	Name          string
	NumPartitions int
	Config        map[string]string
}

type CreateTopicResponse struct {
	Response interface{}
}

type DeleteTopicRequest struct {
	Name           string
	Force          bool //  only required for pulsar and ignored for kafka
	NonPartitioned bool //  only required for pulsar and ignored for kafka
}

type DeleteTopicResponse struct {
	Response interface{}
}

type GetMessagesFromTopicRequest struct {
	NumOfMessages int32
	TimeoutMs     int
}

type GetMessagesFromTopicResponse struct {
	Messages []ReceivedMessage
}

type GetTopicMetadataRequest struct {
	Topic     string
	Partition int32
}

type GetTopicMetadataResponse struct {
	Topic     string
	Partition int32
	Offset    int32
}

type KafkaBroker struct {
	Producer *kafkapkg.Producer
	Consumer *kafkapkg.Consumer
	Admin    *kafkapkg.AdminClient

	// holds the broker config
	Config *BrokerConfig

	// holds the client configs
	POptions *ProducerClientOptions
	COptions *ConsumerClientOptions
	AOptions *AdminClientOptions
	// contains filtered or unexported fields
}

type MessageHeader struct {
	// the unique identifier of each broker message
	MessageID string
	// time at which the message was pushed onto the broker
	PublishTime time.Time
	// the first topic from where the message originated from
	// the message eventually can cycle through multiple delay and dead-letter topics
	SourceTopic string
	// the current retry topic through which the message is read from
	RetryTopic string
	// the subscription name from where this message originated
	Subscription string
	// the number of retries already attempted for the current message
	CurrentRetryCount int32
	// max attempted based on the dead-letter policy configured in the subscription
	MaxRetryCount int32
	// the current topic where the message is being read from
	CurrentTopic string
	// the delay with which retry began
	InitialDelayInterval uint
	// the current delay calculated being after applying backoff
	CurrentDelayInterval uint
	// the actual interval being used from the pre-defined list of intervals
	ClosestDelayInterval uint
	// the dead letter topic name where is message will be pushed after exhausted all retries
	DeadLetterTopic string
	// the time interval after which this message can be read by the delay-consumer on a retry cycle
	NextDeliveryTime time.Time
	// Sequence number for ordered message delivery
	CurrentSequence int32
	// sequence number of the previous message in ordered message delivery
	PrevSequence int32
}

type ModifyTopicConfigRequest struct {
	TopicConfigs []TopicConfig
}

type PartitionOffset struct {
	Partition int32
	Offset    int32
}

type PauseOnTopicRequest struct {
	Topic     string
	Partition int32
}

type Producer interface {
	// SendMessage sends a message on the topic
	SendMessage(context.Context, SendMessageToTopicRequest) (*SendMessageToTopicResponse, error)

	// IsClosed checks if producer has been closed
	IsClosed(context.Context) bool

	// Shutdown closes the producer
	Shutdown(context.Context)

	// Flush flushes the producer buffer
	Flush(timeoutMs int) error
}

type ProducerClientOptions struct {
	Topic     string
	TimeoutMs int64
}

type ProducerConfig struct {
	RetryBackoff       time.Duration
	Partitioner        string
	MaxRetry           int
	MaxMessages        int
	CompressionEnabled bool
	CompressionType    string
	RetryAck           string
}

type PulsarBroker struct {
	Consumer pulsar.Consumer
	Producer pulsar.Producer
	Admin    pulsarctl.Client

	// holds the broker config
	Config *BrokerConfig

	// holds the client configs
	POptions *ProducerClientOptions
	COptions *ConsumerClientOptions
	AOptions *AdminClientOptions
}

type ReceivedMessage struct {
	Data        []byte
	Topic       string
	Partition   int32
	Offset      int32
	OrderingKey string
	Attributes  []map[string][]byte
	MessageHeader
}

type ResumeOnTopicRequest struct {
	Topic     string
	Partition int32
}

type SendMessageToTopicRequest struct {
	Topic       string
	Message     []byte
	OrderingKey string
	TimeoutMs   int
	Attributes  []map[string][]byte
	MessageHeader
}

type SendMessageToTopicResponse struct {
	MessageID string
}

type TopicConfig struct {
	Name   string
	Config map[string]string
}