feat: add pubsub subscribe function

Author: fandujar

app.go

@@ -13,6 +13,8 @@ import (
 )
 
 type AppConfig struct {
+	Name        string
+	Version     string
 	LogLevel    zerolog.Level
 	HealthCheck *HealthCheck
 	WebApp      *WebApp
@@ -43,6 +45,10 @@ func NewApp(config *AppConfig) *App {
 	}
 }
 
+func (app *App) AddWorker(worker *Worker) {
+	app.Worker = worker
+}
+
 func (app *App) Run() error {
 	zerolog.SetGlobalLevel(app.LogLevel)
 
@@ -100,6 +106,13 @@ func (app *App) Run() error {
 		}()
 	}
 
+	if app.Worker != nil {
+		go func() {
+			log.Info().Msg("starting worker")
+			app.Worker.Start()
+		}()
+	}
+
 	<-shutdown
 	log.Info().Msg("shutting down")
 

examples/worker/worker

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/fandujar/golaze"
 )
@@ -14,23 +15,38 @@ func main() {
 	task := golaze.NewTask(
 		&golaze.TaskConfig{
 			Name:        "task run forever",
-			Repeat:      -1,
+			Repeat:      1,
 			RepeatDelay: 10,
+			Timeout:     15 * time.Second,
 			Exec: func(state *golaze.State) error {
 				if state.Get("counter") == nil {
 					state.Set("counter", 0)
 				}
 
 				state.Set("counter", state.Get("counter").(int)+1)
 				fmt.Printf("running task example: %d\n", state.Get("counter"))
-				// time.Sleep(10 * time.Second)
+				time.Sleep(10 * time.Second)
 				return nil
 			},
 		})
 
 	worker.AddTask(task)
 	go worker.Start()
 
+	// print if task is running
+	for {
+		if task.IsRunning() {
+			fmt.Println("task is running")
+		} else {
+			fmt.Println("task is not running")
+		}
+
+		if <-task.Done {
+			fmt.Println("task finished")
+			break
+		}
+	}
+
 	// wait for the worker to finish
 	<-worker.Shutdown
 	fmt.Println("worker finished")

examples/worker_run_forever_task/main.go

@@ -1,24 +1,73 @@
 package gcp
 
+import (
+	"context"
+	"fmt"
+
+	"cloud.google.com/go/pubsub"
+)
+
 type PubSubClientConfig struct {
-	ProjectID string
-	TopicID   string
+	ProjectID      string
+	TopicID        string
+	SubscriptionID string
 }
 
 type PubSubClient struct {
 	*PubSubClientConfig
+	client *pubsub.Client
 }
 
-func NewPubSubClient(config *PubSubClientConfig) *PubSubClient {
+func NewPubSubClient(config *PubSubClientConfig) (*PubSubClient, error) {
+	ctx := context.Background()
+	client, err := pubsub.NewClient(ctx, config.ProjectID)
+	if err != nil {
+		return nil, fmt.Errorf("pubsub.NewClient: %v", err)
+	}
+
 	return &PubSubClient{
 		config,
-	}
+		client,
+	}, nil
 }
 
 func (client *PubSubClient) Publish(message string) error {
 	return nil
 }
 
-func (client *PubSubClient) Subscribe() (chan string, error) {
-	return nil, nil
+func (client *PubSubClient) Subscribe(ctx context.Context) (chan string, error) {
+	sub := client.client.Subscription(client.SubscriptionID)
+
+	// Create the subscription if it doesn't exist.
+	exists, err := sub.Exists(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if !exists {
+		_, err := client.client.CreateSubscription(ctx, client.SubscriptionID, pubsub.SubscriptionConfig{
+			Topic: client.client.Topic(client.TopicID),
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create subscription: %v", err)
+		}
+	}
+
+	// Create a channel to pass messages received from Pub/Sub.
+	msgCh := make(chan string)
+
+	// Start a goroutine to receive messages.
+	go func() {
+		err := sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
+			// Forward message data to channel and then acknowledge the message.
+			msgCh <- string(msg.Data)
+			msg.Ack()
+		})
+		if err != nil {
+			// Handling error, for example logging it. Not closing channel here to avoid panic in case of send on closed channel.
+			fmt.Printf("Error receiving messages: %v\n", err)
+		}
+	}()
+
+	return msgCh, nil
 }

gcp/pubsub.go

@@ -3,9 +3,41 @@ module github.com/fandujar/golaze
 go 1.22.1
 
 require (
+	cloud.google.com/go v0.112.1 // indirect
+	cloud.google.com/go/compute v1.24.0 // indirect
+	cloud.google.com/go/compute/metadata v0.2.3 // indirect
+	cloud.google.com/go/iam v1.1.6 // indirect
+	cloud.google.com/go/pubsub v1.37.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-chi/chi/v5 v5.0.12 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/s2a-go v0.1.7 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.2 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/rs/zerolog v1.32.0 // indirect
-	golang.org/x/sys v0.12.0 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.48.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.48.0 // indirect
+	go.opentelemetry.io/otel v1.23.0 // indirect
+	go.opentelemetry.io/otel/metric v1.23.0 // indirect
+	go.opentelemetry.io/otel/trace v1.23.0 // indirect
+	golang.org/x/crypto v0.19.0 // indirect
+	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/oauth2 v0.17.0 // indirect
+	golang.org/x/sync v0.6.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
+	google.golang.org/api v0.167.0 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
+	google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240304161311-37d4d3c04a78 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240228224816-df926f6c8641 // indirect
+	google.golang.org/grpc v1.62.0 // indirect
+	google.golang.org/protobuf v1.32.0 // indirect
 )