mirror of
https://github.com/chrislusf/seaweedfs
synced 2025-06-29 16:22:46 +02:00
02773a6107
* rename * set agent address * refactor * add agent sub * pub messages * grpc new client * can publish records via agent * send init message with session id * fmt * check cancelled request while waiting * use sessionId * handle possible nil stream * subscriber process messages * separate debug port * use atomic int64 * less logs * minor * skip io.EOF * rename * remove unused * use saved offsets * do not reuse session, since always session id is new after restart remove last active ts from SessionEntry * simplify printing * purge unused * just proxy the subscription, skipping the session step * adjust offset types * subscribe offset type and possible value * start after the known tsns * avoid wrongly set startPosition * move * remove * refactor * typo * fix * fix changed path
69 lines
2.6 KiB
Go
69 lines
2.6 KiB
Go
package sub_client
|
|
|
|
import (
|
|
"context"
|
|
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
|
|
"google.golang.org/grpc"
|
|
"sync"
|
|
)
|
|
|
|
type SubscriberConfiguration struct {
|
|
ClientId string
|
|
ConsumerGroup string
|
|
ConsumerGroupInstanceId string
|
|
GrpcDialOption grpc.DialOption
|
|
MaxPartitionCount int32 // how many partitions to process concurrently
|
|
SlidingWindowSize int32 // how many messages to process concurrently per partition
|
|
}
|
|
|
|
func (s *SubscriberConfiguration) String() string {
|
|
return "ClientId: " + s.ClientId + ", ConsumerGroup: " + s.ConsumerGroup + ", ConsumerGroupInstanceId: " + s.ConsumerGroupInstanceId
|
|
}
|
|
|
|
type ContentConfiguration struct {
|
|
Topic topic.Topic
|
|
Filter string
|
|
PartitionOffsets []*schema_pb.PartitionOffset
|
|
OffsetType schema_pb.OffsetType
|
|
OffsetTsNs int64
|
|
}
|
|
|
|
type OnDataMessageFn func(m *mq_pb.SubscribeMessageResponse_Data)
|
|
type OnCompletionFunc func()
|
|
|
|
type TopicSubscriber struct {
|
|
ctx context.Context
|
|
SubscriberConfig *SubscriberConfiguration
|
|
ContentConfig *ContentConfiguration
|
|
brokerPartitionAssignmentChan chan *mq_pb.SubscriberToSubCoordinatorResponse
|
|
brokerPartitionAssignmentAckChan chan *mq_pb.SubscriberToSubCoordinatorRequest
|
|
OnDataMessageFunc OnDataMessageFn
|
|
OnCompletionFunc OnCompletionFunc
|
|
bootstrapBrokers []string
|
|
activeProcessors map[topic.Partition]*ProcessorState
|
|
activeProcessorsLock sync.Mutex
|
|
PartitionOffsetChan chan KeyedOffset
|
|
}
|
|
|
|
func NewTopicSubscriber(ctx context.Context, bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration, partitionOffsetChan chan KeyedOffset) *TopicSubscriber {
|
|
return &TopicSubscriber{
|
|
ctx: ctx,
|
|
SubscriberConfig: subscriber,
|
|
ContentConfig: content,
|
|
brokerPartitionAssignmentChan: make(chan *mq_pb.SubscriberToSubCoordinatorResponse, 1024),
|
|
brokerPartitionAssignmentAckChan: make(chan *mq_pb.SubscriberToSubCoordinatorRequest, 1024),
|
|
bootstrapBrokers: bootstrapBrokers,
|
|
activeProcessors: make(map[topic.Partition]*ProcessorState),
|
|
PartitionOffsetChan: partitionOffsetChan,
|
|
}
|
|
}
|
|
|
|
func (sub *TopicSubscriber) SetOnDataMessageFn(fn OnDataMessageFn) {
|
|
sub.OnDataMessageFunc = fn
|
|
}
|
|
|
|
func (sub *TopicSubscriber) SetCompletionFunc(onCompletionFn OnCompletionFunc) {
|
|
sub.OnCompletionFunc = onCompletionFn
|
|
}
|