chore: rename pubsub processor struct (#1389)

cre8ivejp · web-flow · commit f333880e64cc · 2025-01-08T16:38:35.000+09:00
Signed-off-by: Alessandro Yuichi Okimoto &lt;yuichijpn@gmail.com&gt;
diff --git a/pkg/subscriber/cmd/server/server.go b/pkg/subscriber/cmd/server/server.go
@@ -319,7 +319,7 @@ func (s *server) Run(ctx context.Context, metrics metrics.Metrics, logger *zap.L
 	}
 	defer batchClient.Close()
 
-	processors, err := s.registerProcessorMap(
+	pubSubProcessors, err := s.registerPubSubProcessorMap(
 		ctx,
 		environmentClient,
 		mysqlClient,
@@ -337,7 +337,7 @@ func (s *server) Run(ctx context.Context, metrics metrics.Metrics, logger *zap.L
 		return err
 	}
 
-	multiPubSub, err := s.startMultiPubSub(ctx, processors, logger)
+	multiPubSub, err := s.startMultiPubSub(ctx, pubSubProcessors, logger)
 	if err != nil {
 		return err
 	}
@@ -394,7 +394,7 @@ func (s *server) createMySQLClient(
 
 func (s *server) startMultiPubSub(
 	ctx context.Context,
-	processors *processor.Processors,
+	processors *processor.PubSubProcessors,
 	logger *zap.Logger,
 ) (*subscriber.MultiSubscriber, error) {
 	multiSubscriber := subscriber.NewMultiSubscriber(
@@ -422,7 +422,7 @@ func (s *server) startMultiPubSub(
 				// we should skip the error, just log it here
 				continue
 			}
-			multiSubscriber.AddSubscriber(subscriber.NewSubscriber(
+			multiSubscriber.AddSubscriber(subscriber.NewPubSubSubscriber(
 				name, config, p,
 				subscriber.WithLogger(logger),
 			))
@@ -461,7 +461,7 @@ func (s *server) startMultiPubSub(
 	return multiSubscriber, nil
 }
 
-func (s *server) registerProcessorMap(
+func (s *server) registerPubSubProcessorMap(
 	ctx context.Context,
 	environmentClient environmentclient.Client,
 	mysqlClient mysql.Client,
@@ -474,8 +474,8 @@ func (s *server) registerProcessorMap(
 	sender notificationsender.Sender,
 	registerer metrics.Registerer,
 	logger *zap.Logger,
-) (*processor.Processors, error) {
-	processors := processor.NewProcessors(registerer)
+) (*processor.PubSubProcessors, error) {
+	processors := processor.NewPubSubProcessors(registerer)
 	writer.RegisterMetrics(registerer)
 
 	processorsConfigBytes, err := os.ReadFile(*s.processorsConfig)
diff --git a/pkg/subscriber/processor/auditlog_persister.go b/pkg/subscriber/processor/auditlog_persister.go
@@ -50,7 +50,7 @@ func NewAuditLogPersister(
 	config interface{},
 	mysqlClient mysql.Client,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	auditLogPersisterJsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("AuditLogPersister: invalid config")
diff --git a/pkg/subscriber/processor/domain_event_informer.go b/pkg/subscriber/processor/domain_event_informer.go
@@ -50,7 +50,7 @@ func NewDomainEventInformer(
 	environmentClient environmentclient.Client,
 	sender sender.Sender,
 	logger *zap.Logger,
-) subscriber.Processor {
+) subscriber.PubSubProcessor {
 	return &domainEventInformer{
 		environmentClient: environmentClient,
 		sender:            sender,
diff --git a/pkg/subscriber/processor/evaluation_events_evaluation_count_event_persister.go b/pkg/subscriber/processor/evaluation_events_evaluation_count_event_persister.go
@@ -70,7 +70,7 @@ func NewEvaluationCountEventPersister(
 	mysqlClient mysql.Client,
 	evaluationCountCacher cache.MultiGetDeleteCountCache,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	evaluationCountEventPersisterJsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("EvaluationCountEventPersister: invalid config")
diff --git a/pkg/subscriber/processor/events_dwh_persister.go b/pkg/subscriber/processor/events_dwh_persister.go
@@ -63,7 +63,7 @@ func NewEventsDWHPersister(
 	ftClient featureclient.Client,
 	persisterName string,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	jsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("eventsDWHPersister: invalid config")
diff --git a/pkg/subscriber/processor/events_ops_persister.go b/pkg/subscriber/processor/events_ops_persister.go
@@ -58,7 +58,7 @@ func NewEventsOPSPersister(
 	ftClient featureclient.Client,
 	persisterName string,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	jsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("eventsOPSPersister: invalid config")
diff --git a/pkg/subscriber/processor/metrics_event_persister.go b/pkg/subscriber/processor/metrics_event_persister.go
@@ -51,7 +51,7 @@ type metricsEventPersister struct {
 func NewMetricsEventPersister(
 	registerer metrics.Registerer,
 	logger *zap.Logger,
-) subscriber.Processor {
+) subscriber.PubSubProcessor {
 	return &metricsEventPersister{
 		storage: storage.NewStorage(logger, registerer),
 		logger:  logger,
diff --git a/pkg/subscriber/processor/processors.go b/pkg/subscriber/processor/processors.go
@@ -39,22 +39,22 @@ var (
 	unsupportedProcessorErr = errors.New("subscriber: unsupported processor")
 )
 
-type Processors struct {
-	processorMap map[string]subscriber.Processor
+type PubSubProcessors struct {
+	processorMap map[string]subscriber.PubSubProcessor
 }
 
-func NewProcessors(r metrics.Registerer) *Processors {
+func NewPubSubProcessors(r metrics.Registerer) *PubSubProcessors {
 	registerMetrics(r)
-	return &Processors{
-		processorMap: make(map[string]subscriber.Processor),
+	return &PubSubProcessors{
+		processorMap: make(map[string]subscriber.PubSubProcessor),
 	}
 }
 
-func (p *Processors) RegisterProcessor(name string, processor subscriber.Processor) {
+func (p *PubSubProcessors) RegisterProcessor(name string, processor subscriber.PubSubProcessor) {
 	p.processorMap[name] = processor
 }
 
-func (p *Processors) GetProcessorByName(name string) (subscriber.Processor, error) {
+func (p *PubSubProcessors) GetProcessorByName(name string) (subscriber.PubSubProcessor, error) {
 	if p, ok := p.processorMap[name]; ok {
 		return p, nil
 	}
diff --git a/pkg/subscriber/processor/push_sender.go b/pkg/subscriber/processor/push_sender.go
@@ -63,7 +63,7 @@ func NewPushSender(
 	batchClient btclient.Client,
 	mysqlClient mysql.Client,
 	logger *zap.Logger,
-) subscriber.Processor {
+) subscriber.PubSubProcessor {
 	return &pushSender{
 		featureClient: featureClient,
 		batchClient:   batchClient,
diff --git a/pkg/subscriber/processor/segment_user_persister.go b/pkg/subscriber/processor/segment_user_persister.go
@@ -65,7 +65,7 @@ func NewSegmentUserPersister(
 	batchClient btclient.Client,
 	mysqlClient mysql.Client,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	segmentPersisterJsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("SegmentUserPersister: invalid config")
diff --git a/pkg/subscriber/processor/user_event_persister.go b/pkg/subscriber/processor/user_event_persister.go
@@ -50,7 +50,7 @@ func NewUserEventPersister(
 	config interface{},
 	mysqlClient mysql.Client,
 	logger *zap.Logger,
-) (subscriber.Processor, error) {
+) (subscriber.PubSubProcessor, error) {
 	userEventPerisiterJsonConfig, ok := config.(map[string]interface{})
 	if !ok {
 		logger.Error("UserEventPersister: invalid config")
diff --git a/pkg/subscriber/subscriber.go b/pkg/subscriber/subscriber.go
@@ -53,12 +53,12 @@ type Subscriber interface {
 	Stop()
 }
 
-type Processor interface {
+type PubSubProcessor interface {
 	Process(ctx context.Context, msgChan <-chan *puller.Message) error
 }
 
 type OnDemandProcessor interface {
-	Processor
+	PubSubProcessor
 	Switch(ctx context.Context) (bool, error)
 }
 
@@ -73,26 +73,26 @@ type Configuration struct {
 	WorkerNum                    int    `json:"workerNum"`
 }
 
-type subscriber struct {
+type pubSubSubscriber struct {
 	name          string
 	configuration Configuration
-	processor     Processor
+	processor     PubSubProcessor
 	cancel        context.CancelFunc
 	opts          options
 	logger        *zap.Logger
 }
 
-func NewSubscriber(
+func NewPubSubSubscriber(
 	name string,
 	configuration Configuration,
-	processor Processor,
+	processor PubSubProcessor,
 	opts ...Option,
 ) Subscriber {
 	options := defaultOptions
 	for _, o := range opts {
 		o(&options)
 	}
-	return &subscriber{
+	return &pubSubSubscriber{
 		name:          name,
 		configuration: configuration,
 		processor:     processor,
@@ -101,7 +101,7 @@ func NewSubscriber(
 	}
 }
 
-func (s subscriber) Run(ctx context.Context) {
+func (s pubSubSubscriber) Run(ctx context.Context) {
 	s.logger.Debug("subscriber starting",
 		zap.String("name", s.name),
 		zap.String("project", s.configuration.Project),
@@ -134,13 +134,13 @@ func (s subscriber) Run(ctx context.Context) {
 		zap.String("name", s.name))
 }
 
-func (s subscriber) Stop() {
+func (s pubSubSubscriber) Stop() {
 	if s.cancel != nil {
 		s.cancel()
 	}
 }
 
-func (s subscriber) createPuller(
+func (s pubSubSubscriber) createPuller(
 	ctx context.Context,
 ) puller.RateLimitedPuller {
 	pubsubClient, err := pubsub.NewClient(

Original file line number	Diff line number	Diff line change
`@@ -39,22 +39,22 @@ var (`
`39`	`39`	`unsupportedProcessorErr = errors.New("subscriber: unsupported processor")`
`40`	`40`	`)`
`41`	`41`
`42`		`-type Processors struct {`
`43`		`- processorMap map[string]subscriber.Processor`
	`42`	`+type PubSubProcessors struct {`
	`43`	`+ processorMap map[string]subscriber.PubSubProcessor`
`44`	`44`	`}`
`45`	`45`
`46`		`-func NewProcessors(r metrics.Registerer) *Processors {`
	`46`	`+func NewPubSubProcessors(r metrics.Registerer) *PubSubProcessors {`
`47`	`47`	`registerMetrics(r)`
`48`		`- return &Processors{`
`49`		`- processorMap: make(map[string]subscriber.Processor),`
	`48`	`+ return &PubSubProcessors{`
	`49`	`+ processorMap: make(map[string]subscriber.PubSubProcessor),`
`50`	`50`	`}`
`51`	`51`	`}`
`52`	`52`
`53`		`-func (p *Processors) RegisterProcessor(name string, processor subscriber.Processor) {`
	`53`	`+func (p *PubSubProcessors) RegisterProcessor(name string, processor subscriber.PubSubProcessor) {`
`54`	`54`	`p.processorMap[name] = processor`
`55`	`55`	`}`
`56`	`56`
`57`		`-func (p *Processors) GetProcessorByName(name string) (subscriber.Processor, error) {`
	`57`	`+func (p *PubSubProcessors) GetProcessorByName(name string) (subscriber.PubSubProcessor, error) {`
`58`	`58`	`if p, ok := p.processorMap[name]; ok {`
`59`	`59`	`return p, nil`
`60`	`60`	`}`