update pubsub

Author: conduongtong

consumer.go

@@ -44,7 +44,7 @@ func ConfigureSubscription(subscription *pubsub.Subscription, c SubscriptionConf
 	return subscription
 }
 
-func (c *Consumer) Consume(ctx context.Context, caller mq.ConsumerCaller) {
+func (c *Consumer) Consume(ctx context.Context, handle func(context.Context, *mq.Message, error) error) {
 	er1 := c.Subscription.Receive(ctx, func(ctx2 context.Context, m *pubsub.Message) {
 		message := mq.Message{
 			Id:         m.ID,
@@ -55,9 +55,9 @@ func (c *Consumer) Consume(ctx context.Context, caller mq.ConsumerCaller) {
 		if c.AckOnConsume {
 			m.Ack()
 		}
-		caller.Call(ctx2, &message, nil)
+		handle(ctx2, &message, nil)
 	})
 	if er1 != nil {
-		caller.Call(ctx, nil, er1)
+		handle(ctx, nil, er1)
 	}
 }

producer.go

@@ -55,13 +55,13 @@ func ConfigureTopic(topic *pubsub.Topic, c TopicConfig) *pubsub.Topic {
 	return topic
 }
 
-func (c *Producer) Produce(ctx context.Context, data []byte, messageAttributes *map[string]string) (string, error) {
+func (c *Producer) Produce(ctx context.Context, data []byte, messageAttributes map[string]string) (string, error) {
 	msg := &pubsub.Message{
 		Data: data,
 	}
 
 	if messageAttributes != nil {
-		msg.Attributes = *messageAttributes
+		msg.Attributes = messageAttributes
 	}
 
 	publishResult := c.Topic.Publish(ctx, msg)

pubsub_health_checker.go

@@ -23,10 +23,24 @@ type PubSubHealthChecker struct {
 	resourceId     string
 }
 
-func NewPubSubHealthChecker(name string, client *pubsub.Client, timeout time.Duration, permissionType PermissionType, resourceId string) *PubSubHealthChecker {
-	return &PubSubHealthChecker{name, client, timeout, permissionType, resourceId}
+func NewPubSubHealthChecker(name string, client *pubsub.Client, resourceId string, permissionType PermissionType, timeout ...time.Duration) *PubSubHealthChecker {
+	if len(timeout) >= 1 {
+		return &PubSubHealthChecker{name: name, client: client, permissionType: permissionType, resourceId: resourceId, timeout: timeout[0]}
+	}
+	return &PubSubHealthChecker{name: name, client: client, permissionType: permissionType, resourceId: resourceId, timeout: 4 * time.Second}
+}
+func NewPubHealthChecker(name string, client *pubsub.Client, resourceId string, timeout ...time.Duration) *PubSubHealthChecker {
+	if len(timeout) >= 1 {
+		return &PubSubHealthChecker{name: name, client: client, permissionType: PermissionPublish, resourceId: resourceId, timeout: timeout[0]}
+	}
+	return &PubSubHealthChecker{name: name, client: client, permissionType: PermissionPublish, resourceId: resourceId, timeout: 4 * time.Second}
+}
+func NewSubHealthChecker(name string, client *pubsub.Client, resourceId string, timeout ...time.Duration) *PubSubHealthChecker {
+	if len(timeout) >= 1 {
+		return &PubSubHealthChecker{name: name, client: client, permissionType: PermissionSubscribe, resourceId: resourceId, timeout: timeout[0]}
+	}
+	return &PubSubHealthChecker{name: name, client: client, permissionType: PermissionSubscribe, resourceId: resourceId, timeout: 4 * time.Second}
 }
-
 func (h *PubSubHealthChecker) Name() string {
 	return h.name
 }