consumer

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

type AMQPOpt struct {
	Exchange            string
	ExchangeType        string
	Heartbeat           time.Duration //心跳检查间隔 s
	Vhost               string
	QueueName           string
	RouterKey           string
	Username            string //鉴权 用户名
	Password            string //鉴权 密码
	AMQPQueueDeclare    mq.AMQPQueueDeclareOPT
	AMQPExchangeDeclare mq.AMQPExchangeDeclareOPT
	AMQPQueueBind       mq.AMQPQueueBindOPT
	AMQPConsumer        mq.AMQPConsumeOPT
}

type Cmessage struct {
	Key  []byte
	Body []byte
	Info map[string]interface{}
}

type Coptions struct {
	Kafka          KafkaOpt
	Nsq            NsqOpt
	AMQP           AMQPOpt
	LookupInterval time.Duration //重连时间
	Address        string
}

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

type KafkaOpt struct {
	Version    *sarama.KafkaVersion
	Topic      []string
	GroupID    string               //offset
	Offsets    []OffsetGroup        //不同group，topic的offset信息
	Partition  mapset.Set           //接受的partition
	TimeFilter map[string]time.Time //时间范围，string只有『After』[Befter](不包含，开区间)
	UserName   string               //鉴权 用户名
	Password   string               //鉴权 密码
}

type MqConsumer interface {
	// RegisterReceiver 注册接受者
	RegisterReceiver(Receiver)
	// Start 开始接受消息
	Start()
	// Stop 结束
	Stop()
}

type NsqConsumer struct {
	MaxWorkerNum int
	// contains filtered or unexported fields
}

type NsqOpt struct {
	Topic      string
	Channel    string
	AuthSecret string //鉴权 secret
	MaxTaskNum int
}

type OffsetGroup struct {
	Topic     string
	Partition int32
	Offset    int64
	Metadata  string
}

type Option func(opt *Coptions)

type Receiver interface {
	OnError(err error)           // 处理遇到的错误，当MQ对象发生了错误，他需要告诉接收者处理错误
	OnReceive(msg Cmessage) bool // 处理收到的消息, 这里需要告知MQ对象消息是否处理成功
	GetTopic() Topic             //消息的topic，queue等信息
}

type Topic struct {
	Topic     string
	Topics    []string
	Queue     string
	RouterKey string
	Channel   string
	Args      map[string]interface{}
}