feat: inject pubsub to event handler

cmd/handler.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/ucpr/mongo-streamer/internal/log"
+	"github.com/ucpr/mongo-streamer/internal/pubsub"
+)
+
+type EventHandler struct {
+	pubsub pubsub.Publisher
+}
+
+func NewEventHandler(ps pubsub.Publisher) *EventHandler {
+	return &EventHandler{
+		pubsub: ps,
+	}
+}
+
+func (e *EventHandler) EventHandler(ctx context.Context, event []byte) error {
+	log.Info("event", slog.String("event", string(event)))
+	return nil
+}

cmd/streamer.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"log/slog"
 	"time"
 
 	"github.com/ucpr/mongo-streamer/internal/config"
@@ -17,13 +16,13 @@ type Streamer struct {
 	st  persistent.StorageBuffer
 }
 
-func NewStreamer(ctx context.Context, cli *mongo.Client, mcfg *config.MongoDB) (*Streamer, error) {
+func NewStreamer(ctx context.Context, cli *mongo.Client, mcfg *config.MongoDB, eh *EventHandler) (*Streamer, error) {
 	stLog := persistent.NewLogWriter()
 	st, err := persistent.NewBuffer(10, 5*time.Second, stLog)
 	if err != nil {
 		return nil, err
 	}
-	cs, err := mongo.NewChangeStream(ctx, cli, mcfg.Database, mcfg.Collection, eventHandler, st)
+	cs, err := mongo.NewChangeStream(ctx, cli, mcfg.Database, mcfg.Collection, eh.EventHandler, st)
 	if err != nil {
 		return nil, err
 	}
@@ -35,11 +34,6 @@ func NewStreamer(ctx context.Context, cli *mongo.Client, mcfg *config.MongoDB) (
 	}, nil
 }
 
-func eventHandler(ctx context.Context, event []byte) error {
-	log.Info("event", slog.String("event", string(event)))
-	return nil
-}
-
 func (s *Streamer) Stream(ctx context.Context) {
 	go func() {
 		s.st.Watch(ctx)

cmd/wire.go

@@ -10,13 +10,16 @@ import (
 	"github.com/ucpr/mongo-streamer/internal/config"
 	"github.com/ucpr/mongo-streamer/internal/http"
 	"github.com/ucpr/mongo-streamer/internal/mongo"
+	"github.com/ucpr/mongo-streamer/internal/pubsub"
 )
 
 func injectStreamer(ctx context.Context) (*Streamer, error) {
 	wire.Build(
 		config.Set,
 		mongo.Set,
+		pubsub.Set,
 		NewStreamer,
+		NewEventHandler,
 	)
 	return nil, nil
 }

internal/pubsub/publisher.go

@@ -5,6 +5,14 @@ import (
 	"time"
 
 	"cloud.google.com/go/pubsub"
+	"github.com/google/wire"
+	"github.com/ucpr/mongo-streamer/internal/config"
+)
+
+//golint:gochecknoglobals
+var Set = wire.NewSet(
+	wire.Bind(new(Publisher), new(*PubSubPublisher)),
+	NewPublisher,
 )
 
 const (
@@ -36,13 +44,13 @@ type PubSubPublisher struct {
 var _ Publisher = (*PubSubPublisher)(nil)
 
 // NewPublisher creates a new publisher.
-func NewPublisher(ctx context.Context, projectID, topicID string) (*PubSubPublisher, error) {
-	cli, err := pubsub.NewClient(ctx, projectID)
+func NewPublisher(ctx context.Context, cfg *config.PubSub) (*PubSubPublisher, error) {
+	cli, err := pubsub.NewClient(ctx, cfg.ProjectID)
 	if err != nil {
 		return nil, err
 	}
 
-	topic := cli.Topic(topicID)
+	topic := cli.Topic(cfg.TopicID)
 	// Set the default values for PublishSettings.
 	topic.PublishSettings.ByteThreshold = publisherByteThreshold
 	topic.PublishSettings.CountThreshold = publisherCountThreshold