Skip to content
Snippets Groups Projects
Select Git revision
  • 8190f92a5f6eb416f138574dc3c42f15dcc54d4c
  • main default protected
  • v1.298.0
  • v1.297.0
  • v1.296.0
  • v1.295.0
  • v1.294.0
  • v1.293.0
  • v1.292.0
  • v1.291.0
  • v1.290.0
  • v1.289.0
  • v1.288.0
  • v1.287.0
  • v1.286.0
  • v1.285.0
  • v1.284.0
  • v1.283.0
  • v1.282.0
  • v1.281.0
  • v1.280.0
  • v1.279.0
22 results

consumer.go

Blame
  • consumer.go 6.52 KiB
    package sqs
    
    import (
    	"context"
    	"encoding/json"
    	"fmt"
    	"math/rand"
    	"os"
    	"path"
    	"reflect"
    	"strings"
    	"time"
    
    	"github.com/aws/aws-lambda-go/events"
    	"github.com/aws/aws-lambda-go/lambda"
    	"github.com/aws/aws-lambda-go/lambdacontext"
    	"github.com/google/uuid"
    	"gitlab.com/uafrica/go-utils/audit"
    	"gitlab.com/uafrica/go-utils/errors"
    	"gitlab.com/uafrica/go-utils/queues"
    	"gitlab.com/uafrica/go-utils/service"
    )
    
    func NewConsumer(requestIDHeaderKey string, routes map[string]interface{}) queues.Consumer {
    	env := os.Getenv("ENVIRONMENT") //todo: support config loading for local dev and env for lambda in prod
    	if env == "" {
    		env = "dev"
    	}
    	router, err := queues.NewRouter(routes)
    	if err != nil {
    		panic(fmt.Sprintf("cannot create router: %+v", err))
    	}
    
    	//legacy message type - when running SQS instance for one type of messages only
    	//when defined, make sure handler exists for this type
    	sqsMessageType := os.Getenv("SQS_MESSAGE_TYPE")
    	if sqsMessageType != "" {
    		if _, err := router.Route(sqsMessageType); err != nil {
    			panic(errors.Errorf("No route defined for SQS_MESSAGE_TYPE=\"%s\"", sqsMessageType))
    		}
    	}
    
    	return consumer{
    		Service: service.New(),
    		//Logger:              logger.New().WithFields(map[string]interface{}{"env": env}),
    		env:                 env,
    		router:              router,
    		requestIDHeaderKey:  requestIDHeaderKey,
    		ConstantMessageType: sqsMessageType,
    		producer:            NewProducer(requestIDHeaderKey),
    		checks:              map[string]queues.ICheck{},
    	}
    }
    
    type consumer struct {
    	service.Service
    	env                 string
    	router              queues.Router
    	requestIDHeaderKey  string
    	ConstantMessageType string //from os.Getenv("SQS_MESSAGE_TYPE")
    	producer            service.Producer
    	checks              map[string]queues.ICheck
    }
    
    //wrap Service.WithStarter to return cron, else cannot be chained
    func (consumer consumer) WithStarter(name string, starter service.IStarter) queues.Consumer {
    	consumer.Service = consumer.Service.WithStarter(name, starter)
    	return consumer
    }