subscriber.go 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. package sub_client
  2. import (
  3. "github.com/seaweedfs/seaweedfs/weed/mq/topic"
  4. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  5. "google.golang.org/grpc"
  6. "sync"
  7. "time"
  8. )
  9. type SubscriberConfiguration struct {
  10. ClientId string
  11. ConsumerGroup string
  12. ConsumerGroupInstanceId string
  13. GrpcDialOption grpc.DialOption
  14. MaxPartitionCount int32 // how many partitions to process concurrently
  15. PerPartitionConcurrency int32 // how many messages to process concurrently per partition
  16. }
  17. type ContentConfiguration struct {
  18. Topic topic.Topic
  19. Filter string
  20. StartTime time.Time
  21. StopTime time.Time
  22. }
  23. type OnEachMessageFunc func(key, value []byte) (err error)
  24. type OnCompletionFunc func()
  25. type TopicSubscriber struct {
  26. SubscriberConfig *SubscriberConfiguration
  27. ContentConfig *ContentConfiguration
  28. brokerPartitionAssignmentChan chan *mq_pb.SubscriberToSubCoordinatorResponse
  29. brokerPartitionAssignmentAckChan chan *mq_pb.SubscriberToSubCoordinatorRequest
  30. OnEachMessageFunc OnEachMessageFunc
  31. OnCompletionFunc OnCompletionFunc
  32. bootstrapBrokers []string
  33. waitForMoreMessage bool
  34. activeProcessors map[topic.Partition]*ProcessorState
  35. activeProcessorsLock sync.Mutex
  36. }
  37. func NewTopicSubscriber(bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration) *TopicSubscriber {
  38. return &TopicSubscriber{
  39. SubscriberConfig: subscriber,
  40. ContentConfig: content,
  41. brokerPartitionAssignmentChan: make(chan *mq_pb.SubscriberToSubCoordinatorResponse, 1024),
  42. brokerPartitionAssignmentAckChan: make(chan *mq_pb.SubscriberToSubCoordinatorRequest, 1024),
  43. bootstrapBrokers: bootstrapBrokers,
  44. waitForMoreMessage: true,
  45. activeProcessors: make(map[topic.Partition]*ProcessorState),
  46. }
  47. }
  48. func (sub *TopicSubscriber) SetEachMessageFunc(onEachMessageFn OnEachMessageFunc) {
  49. sub.OnEachMessageFunc = onEachMessageFn
  50. }
  51. func (sub *TopicSubscriber) SetCompletionFunc(onCompletionFn OnCompletionFunc) {
  52. sub.OnCompletionFunc = onCompletionFn
  53. }