Refactor code

Author: minhduc140583

README.md

@@ -1,15 +1,86 @@
 # pubsub
 
+A fully managed messaging service that allows for event-driven systems and real-time analytics on Google Cloud Platform. Key features include:
+- <b>Scalability</b>: Automatically scales to handle high-throughput workloads.
+- <b>Durability</b>: Ensures message delivery with at-least-once delivery guarantees.
+- <b>Flexibility</b>: Supports both push and pull delivery models.
+- <b>Integration</b>: Easily integrates with other Google Cloud services.
+
+### Use Cases of Google Pub/Sub
+Common use cases include event-driven architectures, log collection, and streaming analytics.
+
+![Microservice Architecture](https://cdn-images-1.medium.com/max/800/1*vKeePO_UC73i7tfymSmYNA.png)
+
+#### Event-Driven Architectures
+- <b>Scenario</b>: Building applications where different components communicate via events (e.g., microservices)
+- <b>Benefit</b>: Decouples components, allowing independent scaling and development
+![A typical micro service](https://cdn-images-1.medium.com/max/800/1*d9kyekAbQYBxH-C6w38XZQ.png)
+#### Log Collection and Monitoring
+- <b>Scenario</b>: Aggregating logs from multiple applications and systems.
+- <b>Benefit</b>: Centralized logging and monitoring, improving visibility and debugging capabilities.
+
+
+#### Streaming analytics
+- <b>Scenario</b>: Collecting and analyzing data streams from various sources like IoT devices, social media, or user activity.
+- <b>Benefit</b>: Enables real-time data processing and analytics, providing timely insights and actions.
+
+### Libraries for Google Pub/Sub
+- GO: [pubsub](https://github.com/core-go/pubsub). Example is at [go-subscription](https://github.com/project-samples/go-subscription)
+- nodejs: [pubsub](https://github.com/core-ts/pubsub). Example is at [pubsub-sample](https://github.com/typescript-tutorial/pubsub-sample)
+
+#### A common flow to consume a message from a message queue
+![A common flow to consume a message from a message queue](https://cdn-images-1.medium.com/max/800/1*Y4QUN6QnfmJgaKigcNHbQA.png)
+- The libraries to implement this flow are:
+  - [mq](https://github.com/core-go/mq) for GOLANG. Example is at [go-subscription](https://github.com/project-samples/go-subscription)
+  - [mq-one](https://www.npmjs.com/package/mq-one) for nodejs. Example is at [pubsub-sample](https://github.com/typescript-tutorial/pubsub-sample)
+
+### Comparison of Google Pub/Sub, Amazon SQS, and Apache Kafka
+#### Google Pub/Sub:
+- <b>Type</b>: Managed real-time messaging service.
+- <b>Use Case</b>: Event-driven architectures, real-time analytics.
+- <b>Scalability</b>: Automatically scales.
+- <b>Delivery Guarantees</b>: At-least-once delivery.
+- <b>Integration</b>: Tight with Google Cloud services.
+- <b>Delivery Models</b>: Push and pull.
+
+#### Amazon SQS
+- <b>Type</b>: Managed message queuing service.
+- <b>Use Case</b>: Decoupling and scaling microservices, asynchronous tasks.
+- <b>Scalability</b>: Automatically scales.
+- <b>Delivery Guarantees</b>: At-least-once, FIFO (exactly-once).
+- <b>Integration</b>: Deep integration with AWS services.
+- <b>Delivery Models</b>: Primarily pull, with long polling.
+
+#### Apache Kafka
+- <b>Type</b>: Open-source event streaming platform.
+- <b>Use Case</b>: High-throughput messaging, event sourcing, log aggregation.
+- <b>Scalability</b>: High with partitioned topics.
+- <b>Delivery Guarantees</b>: Configurable (at-least-once, exactly-once).
+- <b>Integration</b>: Broad ecosystem with various connectors.
+- <b>Delivery Models</b>: Pull-based consumer groups.
+
+### Key Differences
+- <b>Management</b>: Pub/Sub and SQS are managed services, while Kafka is typically self-managed or via managed services like Confluent.
+- <b>Use Case Focus</b>: Pub/Sub and Kafka are ideal for real-time processing, whereas SQS is great for decoupling microservices and handling asynchronous tasks.
+- <b>Delivery Models</b>: Pub/Sub supports push and pull, SQS supports pull with long polling, and Kafka primarily uses pull with consumer groups.
+- <b>Scalability</b>: All three are highly scalable, but Kafka offers the most control over performance tuning.
+- <b>Integration</b>: Pub/Sub integrates well with Google Cloud, SQS with AWS, and Kafka has a broad integration ecosystem.
+
+### When to Use
+- <b>Google Pub/Sub</b>: If you're using Google Cloud and need a managed, real-time messaging solution.
+- <b>Amazon SQS</b>: For reliable, scalable message queuing in AWS environments.
+- <b>Apache Kafka</b>: For complex event streaming and log aggregation, with a need for fine-tuned control and a broad integration ecosystem.
+
 ## Installation
 
-Please make sure to initialize a Go module before installing common-go/pubsub:
+Please make sure to initialize a Go module before installing core-go/pubsub:
 
 ```shell
-go get -u github.com/common-go/pubsub
+go get -u github.com/core-go/pubsub
 ```
 
 Import:
 
 ```go
-import "github.com/common-go/pubsub"
+import "github.com/core-go/pubsub"
 ```

client.go

@@ -5,24 +5,36 @@ import (
 	"context"
 	"fmt"
 	"google.golang.org/api/option"
+	"google.golang.org/api/transport"
 	"log"
 	"os"
 	"reflect"
 	"strconv"
 	"time"
 )
 
-func NewPubSubClientWithRetries(ctx context.Context, projectId string, keyFilename string, retries []time.Duration) (*pubsub.Client, error) {
-	if len(keyFilename) > 0 && existFile(keyFilename) {
-		log.Println("key file exists")
-		c, er1 := pubsub.NewClient(ctx, projectId, option.WithCredentialsFile(keyFilename))
+func NewPubSubClientWithRetries(ctx context.Context, credentials []byte, retries []time.Duration, options ...string) (*pubsub.Client, error) {
+	var projectId string
+	if len(options) > 0 && len(options[0]) > 0 {
+		projectId = options[0]
+	}
+	if credentials != nil && len(credentials) > 0 {
+		opts := option.WithCredentialsJSON(credentials)
+		creds, er0 := transport.Creds(ctx, opts)
+		if er0 != nil {
+			return nil, er0
+		}
+		if len(projectId) == 0 {
+			projectId = creds.ProjectID
+		}
+		c, er1 := pubsub.NewClient(ctx, projectId, opts)
 		if er1 == nil {
 			return c, er1
 		}
 		i := 0
 		err := Retry(retries, func() (err error) {
 			i = i + 1
-			c2, er2 := pubsub.NewClient(ctx, projectId, option.WithCredentialsFile(keyFilename))
+			c2, er2 := pubsub.NewClient(ctx, projectId, opts)
 			if er2 == nil {
 				c = c2
 			}
@@ -33,11 +45,11 @@ func NewPubSubClientWithRetries(ctx context.Context, projectId string, keyFilena
 		}
 		return c, err
 	} else {
-		log.Println("key file doesn't exists")
+		log.Println("empty credentials")
 		return pubsub.NewClient(ctx, projectId)
 	}
 }
-func NewPubSubClient(ctx context.Context, projectId string, keyFilename string) (*pubsub.Client, error) {
+func NewPubSubClientWithFile(ctx context.Context, projectId string, keyFilename string) (*pubsub.Client, error) {
 	if len(keyFilename) > 0 && existFile(keyFilename) {
 		log.Println("key file exists")
 		return pubsub.NewClient(ctx, projectId, option.WithCredentialsFile(keyFilename))
@@ -46,6 +58,28 @@ func NewPubSubClient(ctx context.Context, projectId string, keyFilename string)
 		return pubsub.NewClient(ctx, projectId)
 	}
 }
+func NewPubSubClient(ctx context.Context, credentials []byte, options ...string) (*pubsub.Client, error) {
+	opts := option.WithCredentialsJSON(credentials)
+	var projectId string
+	if len(options) > 0 && len(options[0]) > 0 {
+		projectId = options[0]
+	} else {
+		creds, err := transport.Creds(ctx, opts)
+		projectId = creds.ProjectID
+		if err != nil {
+			panic("Credentials Error: " + err.Error())
+		}
+		if creds == nil {
+			panic("Error: creds is nil")
+		}
+	}
+	if credentials != nil && len(credentials) > 0 {
+		return pubsub.NewClient(ctx, projectId, opts)
+	} else {
+		log.Println("empty credentials")
+		return pubsub.NewClient(ctx, projectId)
+	}
+}
 
 func existFile(filename string) bool {
 	if _, err := os.Stat(filename); err == nil {
@@ -84,17 +118,19 @@ func DurationsFromValue(v interface{}, prefix string, max int) []time.Duration {
 	arr := MakeArray(v, prefix, max)
 	return MakeDurations(arr)
 }
+
 type RetryConfig struct {
-	Retry1  int64 `mapstructure:"1" json:"retry1,omitempty" gorm:"column:retry1" bson:"retry1,omitempty" dynamodbav:"retry1,omitempty" firestore:"retry1,omitempty"`
-	Retry2  int64 `mapstructure:"2" json:"retry2,omitempty" gorm:"column:retry2" bson:"retry2,omitempty" dynamodbav:"retry2,omitempty" firestore:"retry2,omitempty"`
-	Retry3  int64 `mapstructure:"3" json:"retry3,omitempty" gorm:"column:retry3" bson:"retry3,omitempty" dynamodbav:"retry3,omitempty" firestore:"retry3,omitempty"`
-	Retry4  int64 `mapstructure:"4" json:"retry4,omitempty" gorm:"column:retry4" bson:"retry4,omitempty" dynamodbav:"retry4,omitempty" firestore:"retry4,omitempty"`
-	Retry5  int64 `mapstructure:"5" json:"retry5,omitempty" gorm:"column:retry5" bson:"retry5,omitempty" dynamodbav:"retry5,omitempty" firestore:"retry5,omitempty"`
-	Retry6  int64 `mapstructure:"6" json:"retry6,omitempty" gorm:"column:retry6" bson:"retry6,omitempty" dynamodbav:"retry6,omitempty" firestore:"retry6,omitempty"`
-	Retry7  int64 `mapstructure:"7" json:"retry7,omitempty" gorm:"column:retry7" bson:"retry7,omitempty" dynamodbav:"retry7,omitempty" firestore:"retry7,omitempty"`
-	Retry8  int64 `mapstructure:"8" json:"retry8,omitempty" gorm:"column:retry8" bson:"retry8,omitempty" dynamodbav:"retry8,omitempty" firestore:"retry8,omitempty"`
-	Retry9  int64 `mapstructure:"9" json:"retry9,omitempty" gorm:"column:retry9" bson:"retry9,omitempty" dynamodbav:"retry9,omitempty" firestore:"retry9,omitempty"`
+	Retry1 int64 `mapstructure:"1" json:"retry1,omitempty" gorm:"column:retry1" bson:"retry1,omitempty" dynamodbav:"retry1,omitempty" firestore:"retry1,omitempty"`
+	Retry2 int64 `mapstructure:"2" json:"retry2,omitempty" gorm:"column:retry2" bson:"retry2,omitempty" dynamodbav:"retry2,omitempty" firestore:"retry2,omitempty"`
+	Retry3 int64 `mapstructure:"3" json:"retry3,omitempty" gorm:"column:retry3" bson:"retry3,omitempty" dynamodbav:"retry3,omitempty" firestore:"retry3,omitempty"`
+	Retry4 int64 `mapstructure:"4" json:"retry4,omitempty" gorm:"column:retry4" bson:"retry4,omitempty" dynamodbav:"retry4,omitempty" firestore:"retry4,omitempty"`
+	Retry5 int64 `mapstructure:"5" json:"retry5,omitempty" gorm:"column:retry5" bson:"retry5,omitempty" dynamodbav:"retry5,omitempty" firestore:"retry5,omitempty"`
+	Retry6 int64 `mapstructure:"6" json:"retry6,omitempty" gorm:"column:retry6" bson:"retry6,omitempty" dynamodbav:"retry6,omitempty" firestore:"retry6,omitempty"`
+	Retry7 int64 `mapstructure:"7" json:"retry7,omitempty" gorm:"column:retry7" bson:"retry7,omitempty" dynamodbav:"retry7,omitempty" firestore:"retry7,omitempty"`
+	Retry8 int64 `mapstructure:"8" json:"retry8,omitempty" gorm:"column:retry8" bson:"retry8,omitempty" dynamodbav:"retry8,omitempty" firestore:"retry8,omitempty"`
+	Retry9 int64 `mapstructure:"9" json:"retry9,omitempty" gorm:"column:retry9" bson:"retry9,omitempty" dynamodbav:"retry9,omitempty" firestore:"retry9,omitempty"`
 }
+
 func Retry(sleeps []time.Duration, f func() error) (err error) {
 	attempts := len(sleeps)
 	for i := 0; ; i++ {

client_config.go

@@ -1,6 +1,7 @@
 package pubsub
 
 type ClientConfig struct {
-	ProjectId   string `mapstructure:"project_id" json:"projectId,omitempty" gorm:"column:projectid" bson:"projectId,omitempty" dynamodbav:"projectId,omitempty" firestore:"projectId,omitempty"`
-	KeyFilename string `mapstructure:"key_filename" json:"keyFilename,omitempty" gorm:"column:keyfilename" bson:"keyFilename,omitempty" dynamodbav:"keyFilename,omitempty" firestore:"keyFilename,omitempty"`
+	ProjectId   string `yaml:"project_id" mapstructure:"project_id" json:"projectId,omitempty" gorm:"column:projectid" bson:"projectId,omitempty" dynamodbav:"projectId,omitempty" firestore:"projectId,omitempty"`
+	Credentials string `yaml:"credentials" mapstructure:"credentials" json:"credentials,omitempty" gorm:"column:credentials" bson:"credentials,omitempty" dynamodbav:"credentials,omitempty" firestore:"credentials,omitempty"`
+	KeyFilename string `yaml:"key_filename" mapstructure:"key_filename" json:"keyFilename,omitempty" gorm:"column:keyfilename" bson:"keyFilename,omitempty" dynamodbav:"keyFilename,omitempty" firestore:"keyFilename,omitempty"`
 }

health_checker.go

@@ -1,11 +1,10 @@
 package pubsub
 
 import (
+	"cloud.google.com/go/pubsub"
 	"context"
 	"fmt"
 	"time"
-
-	"cloud.google.com/go/pubsub"
 )
 
 type PermissionType int
@@ -70,6 +69,9 @@ func (h *HealthChecker) Build(ctx context.Context, data map[string]interface{},
 	if err == nil {
 		return data
 	}
+	if data == nil {
+		data = make(map[string]interface{}, 0)
+	}
 	data["error"] = err.Error()
 	return data
 }

publisher.go

@@ -1,70 +1,86 @@
 package pubsub
 
 import (
+	"cloud.google.com/go/iam"
+	"cloud.google.com/go/pubsub"
 	"context"
 	"log"
 	"time"
-
-	"cloud.google.com/go/iam"
-	"cloud.google.com/go/pubsub"
 )
 
 var CheckTopicPermission = CheckPermission
 
 type Publisher struct {
-	Client *pubsub.Client
-	Topic  *pubsub.Topic
+	Client  *pubsub.Client
+	Topic   *pubsub.Topic
+	Convert func(context.Context, []byte) ([]byte, error)
 }
 
-func NewPublisher(ctx context.Context, client *pubsub.Client, topicId string, c TopicConfig) *Publisher {
+func NewPublisher(ctx context.Context, client *pubsub.Client, topicId string, c *TopicConfig, options ...func(context.Context, []byte) ([]byte, error)) *Publisher {
 	topic := client.Topic(topicId)
 	CheckTopicPermission(ctx, topic.IAM(), "pubsub.topics.publish")
-	return &Publisher{Client: client, Topic: ConfigureTopic(topic, c)}
+	var convert func(context.Context, []byte) ([]byte, error)
+	if len(options) > 0 {
+		convert = options[0]
+	}
+	return &Publisher{Client: client, Topic: ConfigureTopic(topic, c), Convert: convert}
 }
 
-func NewPublisherByConfig(ctx context.Context, c PublisherConfig) (*Publisher, error) {
+func NewPublisherByConfig(ctx context.Context, c PublisherConfig, options ...func(context.Context, []byte) ([]byte, error)) (*Publisher, error) {
 	if c.Retry.Retry1 <= 0 {
-		client, err := NewPubSubClient(ctx, c.Client.ProjectId, c.Client.KeyFilename)
+		client, err := NewPubSubClient(ctx, []byte(c.Client.Credentials), c.Client.ProjectId)
 		if err != nil {
 			return nil, err
 		}
-		return NewPublisher(ctx, client, c.TopicId, c.Topic), nil
+		return NewPublisher(ctx, client, c.TopicId, c.Topic, options...), nil
 	} else {
 		durations := DurationsFromValue(c.Retry, "Retry", 9)
-		client, err := NewPubSubClientWithRetries(ctx, c.Client.ProjectId, c.Client.KeyFilename, durations)
+		client, err := NewPubSubClientWithRetries(ctx, []byte(c.Client.Credentials), durations, c.Client.ProjectId)
 		if err != nil {
 			return nil, err
 		}
-		return NewPublisher(ctx, client, c.TopicId, c.Topic), nil
+		return NewPublisher(ctx, client, c.TopicId, c.Topic, options...), nil
 	}
 }
 
-func ConfigureTopic(topic *pubsub.Topic, c TopicConfig) *pubsub.Topic {
-	if c.CountThreshold > 0 {
-		topic.PublishSettings.DelayThreshold = time.Duration(c.CountThreshold) * time.Millisecond
-	}
-	if c.DelayThreshold > 0 {
-		topic.PublishSettings.CountThreshold = c.DelayThreshold
-	}
-	if c.ByteThreshold > 0 {
-		topic.PublishSettings.ByteThreshold = c.ByteThreshold
-	}
-	if c.NumGoroutines > 0 {
-		topic.PublishSettings.NumGoroutines = c.NumGoroutines
+func ConfigureTopic(topic *pubsub.Topic, c *TopicConfig) *pubsub.Topic {
+	if c != nil {
+		if c.CountThreshold > 0 {
+			topic.PublishSettings.DelayThreshold = time.Duration(c.CountThreshold) * time.Millisecond
+		}
+		if c.DelayThreshold > 0 {
+			topic.PublishSettings.CountThreshold = c.DelayThreshold
+		}
+		if c.ByteThreshold > 0 {
+			topic.PublishSettings.ByteThreshold = c.ByteThreshold
+		}
+		if c.NumGoroutines > 0 {
+			topic.PublishSettings.NumGoroutines = c.NumGoroutines
+		}
 	}
 	return topic
 }
-
-func (c *Publisher) Publish(ctx context.Context, data []byte, attributes map[string]string) (string, error) {
-	msg := &pubsub.Message{
-		Data: data,
+func (p *Publisher) Publish(ctx context.Context, data []byte, attributes map[string]string) error {
+	msg := &pubsub.Message{Data: data}
+	if attributes != nil {
+		msg.Attributes = attributes
 	}
-
+	publishResult := p.Topic.Publish(ctx, msg)
+	_, err := publishResult.Get(ctx)
+	return err
+}
+func (p *Publisher) PublishData(ctx context.Context, data []byte) error {
+	msg := &pubsub.Message{Data: data}
+	publishResult := p.Topic.Publish(ctx, msg)
+	_, err := publishResult.Get(ctx)
+	return err
+}
+func (p *Publisher) PublishMessage(ctx context.Context, data []byte, attributes map[string]string) (string, error) {
+	msg := &pubsub.Message{Data: data}
 	if attributes != nil {
 		msg.Attributes = attributes
 	}
-
-	publishResult := c.Topic.Publish(ctx, msg)
+	publishResult := p.Topic.Publish(ctx, msg)
 	return publishResult.Get(ctx)
 }
 

publisher_config.go

@@ -1,15 +1,15 @@
 package pubsub
 
 type PublisherConfig struct {
-	TopicId string       `mapstructure:"topic_id" json:"topicId,omitempty" gorm:"column:topicid" bson:"topicId,omitempty" dynamodbav:"topicId,omitempty" firestore:"topicId,omitempty"`
-	Client  ClientConfig `mapstructure:"client" json:"client,omitempty" gorm:"column:client" bson:"client,omitempty" dynamodbav:"client,omitempty" firestore:"client,omitempty"`
-	Topic   TopicConfig  `mapstructure:"topic" json:"topic,omitempty" gorm:"column:topic" bson:"topic,omitempty" dynamodbav:"topic,omitempty" firestore:"topic,omitempty"`
-	Retry   RetryConfig  `mapstructure:"retry" json:"retry,omitempty" gorm:"column:retry" bson:"retry,omitempty" dynamodbav:"retry,omitempty" firestore:"retry,omitempty"`
+	TopicId string       `yaml:"topic_id" mapstructure:"topic_id" json:"topicId,omitempty" gorm:"column:topicid" bson:"topicId,omitempty" dynamodbav:"topicId,omitempty" firestore:"topicId,omitempty"`
+	Client  ClientConfig `yaml:"client" mapstructure:"client" json:"client,omitempty" gorm:"column:client" bson:"client,omitempty" dynamodbav:"client,omitempty" firestore:"client,omitempty"`
+	Topic   *TopicConfig `yaml:"topic" mapstructure:"topic" json:"topic,omitempty" gorm:"column:topic" bson:"topic,omitempty" dynamodbav:"topic,omitempty" firestore:"topic,omitempty"`
+	Retry   RetryConfig  `yaml:"retry" mapstructure:"retry" json:"retry,omitempty" gorm:"column:retry" bson:"retry,omitempty" dynamodbav:"retry,omitempty" firestore:"retry,omitempty"`
 }
 
 type TopicConfig struct {
-	DelayThreshold int `mapstructure:"delay_threshold" json:"delayThreshold,omitempty" gorm:"column:delaythreshold" bson:"delayThreshold,omitempty" dynamodbav:"delayThreshold,omitempty" firestore:"delayThreshold,omitempty"` // MaxMessages
-	CountThreshold int `mapstructure:"count_threshold" json:"countThreshold,omitempty" gorm:"column:countthreshold" bson:"countThreshold,omitempty" dynamodbav:"countThreshold,omitempty" firestore:"countThreshold,omitempty"` // MaxMilliseconds
-	ByteThreshold  int `mapstructure:"byte_threshold" json:"byteThreshold,omitempty" gorm:"column:bytethreshold" bson:"byteThreshold,omitempty" dynamodbav:"byteThreshold,omitempty" firestore:"byteThreshold,omitempty"`  // MaxBytes
-	NumGoroutines  int `mapstructure:"num_goroutines" json:"numGoroutines,omitempty" gorm:"column:numgoroutines" bson:"numGoroutines,omitempty" dynamodbav:"numGoroutines,omitempty" firestore:"numGoroutines,omitempty"`
+	DelayThreshold int `yaml:"delay_threshold" mapstructure:"delay_threshold" json:"delayThreshold,omitempty" gorm:"column:delaythreshold" bson:"delayThreshold,omitempty" dynamodbav:"delayThreshold,omitempty" firestore:"delayThreshold,omitempty"` // MaxMessages
+	CountThreshold int `yaml:"count_threshold" mapstructure:"" json:"countThreshold,omitempty" gorm:"column:countthreshold" bson:"countThreshold,omitempty" dynamodbav:"countThreshold,omitempty" firestore:"countThreshold,omitempty"`                // MaxMilliseconds
+	ByteThreshold  int `yaml:"byte_threshold" mapstructure:"byte_threshold" json:"byteThreshold,omitempty" gorm:"column:bytethreshold" bson:"byteThreshold,omitempty" dynamodbav:"byteThreshold,omitempty" firestore:"byteThreshold,omitempty"`        // MaxBytes
+	NumGoroutines  int `yaml:"num_goroutines" mapstructure:"num_goroutines" json:"numGoroutines,omitempty" gorm:"column:numgoroutines" bson:"numGoroutines,omitempty" dynamodbav:"numGoroutines,omitempty" firestore:"numGoroutines,omitempty"`
 }