add error and debug logs for google and mqtt (#399)

Author: Vipul Rawat

pkg/gofr/datasource/pubsub/google/google.go

@@ -70,6 +70,8 @@ func (g *googleClient) Publish(ctx context.Context, topic string, message []byte
 
 	t, err := g.getTopic(ctx, topic)
 	if err != nil {
+		g.logger.Errorf("error creating %s err: %v", topic, err)
+
 		return err
 	}
 
@@ -80,6 +82,8 @@ func (g *googleClient) Publish(ctx context.Context, topic string, message []byte
 
 	_, err = result.Get(ctx)
 	if err != nil {
+		g.logger.Errorf("error publishing to google topic %s err: %v", topic, err)
+
 		return err
 	}
 
@@ -120,7 +124,7 @@ func (g *googleClient) Subscribe(ctx context.Context, topic string) (*pubsub.Mes
 	})
 
 	if err != nil {
-		g.logger.Errorf("Error getting a message: %s", err.Error())
+		g.logger.Errorf("error getting a message from google: %s", err.Error())
 
 		return nil, err
 	}

pkg/gofr/datasource/pubsub/google/google_test.go

@@ -93,7 +93,7 @@ func TestGoogleClient_PublishTopic_Error(t *testing.T) {
 	g := &googleClient{client: getGoogleClient(t), Config: Config{
 		ProjectID:        "test",
 		SubscriptionName: "sub",
-	}, metrics: mockMetrics}
+	}, metrics: mockMetrics, logger: testutil.NewMockLogger(testutil.DEBUGLOG)}
 	defer g.client.Close()
 
 	ctx, cancel := context.WithCancel(context.Background())

pkg/gofr/datasource/pubsub/mqtt/mqtt.go

@@ -63,14 +63,14 @@ func New(config *Config, logger Logger, metrics Metrics) *MQTT {
 	client := mqtt.NewClient(options)
 
 	if token := client.Connect(); token.Wait() && token.Error() != nil {
-		logger.Errorf("cannot connect to MQTT, HostName : %v, Port : %v, error : %v", config.Hostname, config.Port, token.Error())
+		logger.Errorf("cannot connect to MQTT, host: %v, port: %v, error: %v", config.Hostname, config.Port, token.Error())
 
 		return &MQTT{Client: client, config: config, logger: logger}
 	}
 
 	msg := make(map[string]chan *pubsub.Message)
 
-	logger.Debugf("connected to MQTT, HostName : %v, Port : %v", config.Hostname, config.Port)
+	logger.Debugf("connected to MQTT, host: %v, port: %v", config.Hostname, config.Port)
 
 	return &MQTT{Client: client, config: config, logger: logger, msgChanMap: msg, mu: new(sync.RWMutex), metrics: metrics}
 }
@@ -82,15 +82,13 @@ func getDefaultClient(config *Config, logger Logger, metrics Metrics) *MQTT {
 		clientID = getClientID(config.ClientID)
 	)
 
-	logger.Debugf("using %v clientID for this session", clientID)
-
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(fmt.Sprintf("tcp://%s:%d", host, port))
 	opts.SetClientID(clientID)
 	client := mqtt.NewClient(opts)
 
 	if token := client.Connect(); token.Wait() && token.Error() != nil {
-		logger.Errorf("cannot connect to MQTT, HostName : %v, Port : %v, error : %v", host, port, token.Error())
+		logger.Errorf("cannot connect to MQTT, host: %v, port: %v, error: %v", host, port, token.Error())
 
 		return &MQTT{Client: client, config: config, logger: logger}
 	}
@@ -102,6 +100,7 @@ func getDefaultClient(config *Config, logger Logger, metrics Metrics) *MQTT {
 	msg := make(map[string]chan *pubsub.Message)
 
 	logger.Debugf("connected to MQTT, HostName : %v, Port : %v", config.Hostname, config.Port)
+	logger.Debugf("using %v clientID for this MQTT session", clientID)
 
 	return &MQTT{Client: client, config: config, logger: logger, msgChanMap: msg, mu: new(sync.RWMutex), metrics: metrics}
 }
@@ -174,9 +173,13 @@ func (m *MQTT) Subscribe(ctx context.Context, topic string) (*pubsub.Message, er
 	token := m.Client.Subscribe(topic, m.config.QoS, handler)
 
 	if token.Wait() && token.Error() != nil {
+		m.logger.Errorf("error getting a message from MQTT, err: %v", token.Error())
+
 		return nil, token.Error()
 	}
 
+	m.logger.Debugf("received mqtt message %v on topic %v", string(messg.Value), topic)
+
 	m.metrics.IncrementCounter(ctx, "app_pubsub_subscribe_success_count", "topic", topic)
 
 	// blocks if there are no messages in the channel
@@ -196,6 +199,8 @@ func (m *MQTT) Publish(ctx context.Context, topic string, message []byte) error
 		return token.Error()
 	}
 
+	m.logger.Debugf("published  message %v on topic %v", string(message), topic)
+
 	m.metrics.IncrementCounter(ctx, "app_pubsub_publish_success_count", "topic", topic)
 
 	return nil