Handling messages concurrently

Author: vnalawad-tibco

pulsar/trigger/subscriber/trigger.go

@@ -21,6 +21,7 @@ func init() {
 type Trigger struct {
 	client   pulsar.Client
 	handlers []*Handler
+	logger   log.Logger
 }
 type Handler struct {
 	handler  trigger.Handler
@@ -31,8 +32,6 @@ type Handler struct {
 type Factory struct {
 }
 
-var logger log.Logger
-
 func (*Factory) New(config *trigger.Config) (trigger.Trigger, error) {
 	s := &Settings{}
 	err := metadata.MapToStruct(config.Settings, s, true)
@@ -57,7 +56,7 @@ func (t *Trigger) Metadata() *trigger.Metadata {
 }
 
 func (t *Trigger) Initialize(ctx trigger.InitContext) error {
-	logger = ctx.Logger()
+	t.logger = ctx.Logger()
 	// Init handlers
 	for _, handler := range ctx.GetHandlers() {
 
@@ -95,6 +94,7 @@ func (t *Trigger) Initialize(ctx trigger.InitContext) error {
 			consumeroptions.SubscriptionInitialPosition = pulsar.SubscriptionPositionEarliest
 		}
 
+		consumeroptions.MessageChannel = make(chan pulsar.ConsumerMessage)
 		consumer, err := t.client.Subscribe(consumeroptions)
 		if err != nil {
 			return err
@@ -107,66 +107,89 @@ func (t *Trigger) Initialize(ctx trigger.InitContext) error {
 
 // Start implements util.Managed.Start
 func (t *Trigger) Start() error {
+	t.logger.Info("Starting Trigger")
 	for _, handler := range t.handlers {
 		go handler.consume()
 	}
+	t.logger.Info("Trigger Started")
 	return nil
 }
 
 // Stop implements util.Managed.Stop
 func (t *Trigger) Stop() error {
-	logger.Info("Stopping Trigger")
+	t.logger.Info("Stopping Trigger")
 	for _, handler := range t.handlers {
 		// Stop polling
 		handler.done <- true
 		handler.consumer.Close()
 	}
+	t.logger.Info("Trigger Stopped")
+	return nil
+}
+
+func (t *Trigger) Resume() error {
+	t.logger.Info("Resuming Trigger")
+	err := t.Start()
+	if err == nil {
+		t.logger.Info("Trigger Resumed")
+	}
+	return err
+}
+
+func (t *Trigger) Pause() error {
+	for _, handler := range t.handlers {
+		handler.done <- true
+	}
+	t.logger.Info("Trigger Paused")
 	return nil
 }
 
 func (handler *Handler) consume() {
+	handler.handler.Logger().Info("Pulsar Message consumer is started")
+	defer handler.handler.Logger().Info("Pulsar Message consumer is stopped")
 	for {
 		select {
 		case msg, ok := <-handler.consumer.Chan():
 			if !ok {
-				logger.Error("Error while recieveing message")
-				time.Sleep(5 * time.Second)
+				handler.handler.Logger().Error("Error while receiving message")
+				time.Sleep(1 * time.Second)
 				continue
 			}
-			out := &Output{}
-			if handler.handler.Settings()["format"] != nil &&
-				handler.handler.Settings()["format"].(string) == "JSON" {
-				var obj interface{}
-				err := json.Unmarshal(msg.Payload(), &obj)
-				if err != nil {
-					logger.Errorf("Pulsar consumer, configured to receive JSON formatted messages, was unable to parse message: [%v]", msg.Payload())
-					handler.consumer.Nack(msg)
-					time.Sleep(5 * time.Second)
-					continue
-				} else {
-					out.Payload = obj
-				}
-			} else {
-				out.Payload = string(msg.Payload())
-			}
-			out.Properties = msg.Properties()
-			out.Topic = msg.Topic()
-			logger.Debugf("Message recieved [%v]", out.Payload)
-			// Do something with the message
-			_, err := handler.handler.Handle(context.Background(), out)
-			if err == nil {
-				// Message processed successfully
-				handler.consumer.Ack(msg)
-			} else {
-				// Failed to process messages
-				handler.consumer.Nack(msg)
-				time.Sleep(2 * time.Second)
-			}
+			// Handle messages concurrently on separate goroutine
+			go handler.handleMessage(msg)
 		case <-handler.done:
-			logger.Info("Stopping Message Consumer")
 			return
 		}
-
 	}
+}
 
+func (handler *Handler) handleMessage(msg pulsar.ConsumerMessage) {
+	handler.handler.Logger().Debugf("Message received - %s", string(msg.ID().Serialize()))
+	out := &Output{}
+	if handler.handler.Settings()["format"] != nil &&
+		handler.handler.Settings()["format"].(string) == "JSON" {
+		var obj interface{}
+		err := json.Unmarshal(msg.Payload(), &obj)
+		if err != nil {
+			handler.handler.Logger().Errorf("Pulsar consumer, configured to receive JSON formatted messages, was unable to parse message: [%v]", msg.Payload())
+			handler.consumer.Nack(msg)
+			return
+		} else {
+			out.Payload = obj
+		}
+	} else {
+		out.Payload = string(msg.Payload())
+	}
+	out.Properties = msg.Properties()
+	out.Topic = msg.Topic()
+	handler.handler.Logger().Debugf("Message received [%v]", out.Payload)
+	// Do something with the message
+	_, err := handler.handler.Handle(context.Background(), out)
+	if err == nil {
+		// Message processed successfully
+		handler.consumer.Ack(msg)
+	} else {
+		// Failed to process messages
+		handler.consumer.Nack(msg)
+	}
 }