Main update

Author: Pierre

.gitignore

@@ -0,0 +1 @@
+private_key.json

client/client.go

@@ -0,0 +1,17 @@
+package client
+
+import (
+	"cloud.google.com/go/pubsub"
+	"context"
+	"google.golang.org/api/option"
+	"log"
+)
+
+func InitPubSubClient(ctx context.Context, projectID, credFile string) (*pubsub.Client, error) {
+	client, err := pubsub.NewClient(ctx, projectID, option.WithCredentialsFile(credFile))
+	if err != nil {
+		log.Println("ERROR : client.InitPubSubClient : " + err.Error())
+		return nil, err
+	}
+	return client, nil
+}

consumer/consumer.go

@@ -13,24 +13,25 @@ func InitBuffer(bufferSize int) chan string {
 	return buffer
 }
 
-func Pull(projectID, subName, topicID, credFile string, buffer chan string) {
-	ctx := context.Background()
-	client, err := pubsub.NewClient(ctx, projectID, option.WithCredentialsFile(credFile))
+func Pull(ctx context.Context, pubsubClient *pubsub.Client, subName, topicID string, buffer chan string) error {
 	if err != nil {
-		log.Println("pubsub.NewClient: %v", err)
+		log.Println("ERROR : consumer.Pull : " + err.Error())
+		return err
 	}
 
-	topic := client.Topic(topicID)
+	topic := pubsubClient.Topic(topicID)
 
 	// Create topic subscription if it does not yet exist.
-	sub := client.Subscription(subName)
+	sub := pubsubClient.Subscription(subName)
 	exists, err := sub.Exists(ctx)
 	if err != nil {
-		log.Println("Error checking for subscription: %v", err)
+		log.Println("ERROR : consumer.Pull : " + err.Error())
+		return err
 	}
 	if !exists {
-		if _, err = client.CreateSubscription(ctx, subName, pubsub.SubscriptionConfig{Topic: topic}); err != nil {
-			log.Println("Failed to create subscription: %v", err)
+		if _, err = pubsubClient.CreateSubscription(ctx, subName, pubsub.SubscriptionConfig{Topic: topic}); err != nil {
+			log.Println("ERROR : consumer.Pull : " + err.Error())
+			return err
 		}
 	}
 
@@ -40,6 +41,8 @@ func Pull(projectID, subName, topicID, credFile string, buffer chan string) {
 		msg.Ack()
 	})
 	if err != nil {
-		log.Println("ERROR : pull : %v", err)
+		log.Println("ERROR : consumer.Pull : " + err.Error())
+		return err
 	}
+	return nil
 }

example-consumer.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"github.com/PierreKieffer/pubsubToolBox/client"
 	"github.com/PierreKieffer/pubsubToolBox/consumer"
 	"log"
 )
@@ -9,14 +10,22 @@ var exit = make(chan bool)
 
 func main() {
 
+	ctx := context.Background()
+
+	projectID := "ai-datalake"
+	topicID := "testTopic"
+	subscriberName := "ludo-test"
+
+	pubsubClient, err := client.InitPubSubClient(ctx, projectID, "private_key.json")
+
 	// Init message buffer to receive pulled messages
 	var buffer = consumer.InitBuffer(10)
 
 	// Launch local buffer consumer to process messages
 	go ProcessBuffer(buffer)
 
 	// Launch the pubsub consumer to pull messages
-	go consumer.Pull("PROJECT_ID", "SUBSCRIBER_NAME", "TOPIC_NAME", "private_key.json", buffer)
+	go consumer.Pull(ctx, pubsubClient, subscriberName, topicID, buffer)
 
 	<-exit
 

example-producer.go

@@ -1,12 +1,24 @@
 package main
 
 import (
+	"context"
+	"github.com/PierreKieffer/pubsubToolBox/client"
 	"github.com/PierreKieffer/pubsubToolBox/producer"
+	"log"
 )
 
 func main() {
+	ctx := context.Background()
+
+	projectID := "ai-datalake"
+	topicID := "testTopic"
+
+	pubsubClient, err := client.InitPubSubClient(ctx, projectID, "private_key.json")
+	if err != nil {
+		log.Println(err)
+	}
 
 	message := `{"Message" : "Hello world"}`
-	producer.Publish("PROJECT_NAME", "TOPIC_NAME", "private_key.json", message)
+	producer.Publish(ctx, pubsubClient, topicID, message)
 
 }

producer/producer.go

@@ -0,0 +1,27 @@
+package publisher
+
+import (
+	"context"
+	"google.golang.org/api/option"
+	"log"
+
+	"cloud.google.com/go/pubsub"
+)
+
+func Publish(ctx context.Context, pubsubClient *pubsub.Client, topicID, message string) error {
+
+	t := pubsubClient.Topic(topicID)
+
+	result := t.Publish(ctx, &pubsub.Message{
+		Data: []byte(message),
+	})
+
+	id, err := result.Get(ctx)
+	if err != nil {
+		log.Println("ERROR : publisher.Publish : " + err.Error())
+		return err
+	}
+
+	log.Println("INFO : publisher.Publish : Message published : " + id)
+	return nil
+}