mq

type Codecer interface {
	Marshal(interface{}) ([]byte, error)
	Unmarshal([]byte, interface{}) error
	String() string
}

type Event interface {
	// Topic return the topic of the message
	Topic() string
	// Message return the message body
	Message() *Message
	// ACK message reply operation
	Ack() error
	// Error get the error of message consumption
	Error() error
	// Extra the important information other than the message body
	Extra() map[string]interface{}
}

type Handler func(Event) error

type JsonCodec struct{}

type Logger interface {
	Info(args ...interface{})
	Warn(args ...interface{})
	Error(args ...interface{})
	Errorf(format string, args ...interface{})
	Infof(format string, args ...interface{})
}

type MQ interface {
	Init(...Option) error
	Options() Options
	Address() string
	Connect() error
	Disconnect() error
	Publish(topic string, m *Message, opts ...PublishOption) error
	Subscribe(h Handler, topics []string, opts ...SubscribeOption) (Subscriber, error)
	String() string
}

type MQConfig struct {
	// Addresses of the mq cluster
	Addresses []string `json:"addresses,omitempty"`

	TLSConfig
}

type Message struct {
	Header map[string]string
	Body   []byte
	// contains filtered or unexported fields
}

type Option func(*Options)

type Options struct {
	Addresses []string
	Version   sarama.KafkaVersion
	Secure    bool
	Codec     Codecer
	Username  string
	Password  string
	Algorithm string

	// Handler executed when error happens in mq message processing
	ErrorHandler Handler

	TLSConfig *tls.Config

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context

	Log Logger
	// Whether otel tracing is enabled
	Otel bool
}

type PublishOption func(*PublishOptions)

type PublishOptions struct {
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type Strategy interface {
	Strategy() string
}

type SubscribeOption func(*SubscribeOptions)

type SubscribeOptions struct {
	// AutoAck defaults to true. When a handler returns
	// with a nil error the message is receipt already.
	AutoAck bool

	// Subscribers with the same queue name
	// will create a shared subscription where each
	// receives a subset of messages.
	Queue string

	// RetryNum specifies the one that retry when handle failed
	RetryNum int

	// Strategy specifies the one for handling message
	Strategy Strategy

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type Subscriber interface {
	Options() SubscribeOptions
	Topics() []string
	Unsubscribe() error
}

type TLSConfig struct {
	// CertFile the optional certificate file for client authentication
	CertFile string `json:"cert_file,omitempty"`

	// KeyFile the optional key file for client authentication
	KeyFile string `json:"key_file,omitempty"`

	// CAFile  the optional certificate authority file for TLS client authentication
	CAFile string `json:"ca_file,omitempty"`

	// VerifySSL optional verify ssl certificates chain
	VerifySSL bool `json:"verify_ssl,omitempty"`
}