cloud-services/services/attendant/server/server_consumer.go

package server

import (
	"github.com/fiskerinc/cloud-services/services/attendant/controllers"
	"github.com/fiskerinc/cloud-services/services/attendant/handlers"
	"github.com/fiskerinc/cloud-services/services/attendant/services"

	"github.com/fiskerinc/cloud-services/pkg/common"
	"github.com/fiskerinc/cloud-services/pkg/grpc/kafka_grpc"
	"github.com/fiskerinc/cloud-services/pkg/grpc/sms"
	"github.com/fiskerinc/cloud-services/pkg/kafka"
	"github.com/fiskerinc/cloud-services/pkg/logger"
	"github.com/fiskerinc/cloud-services/pkg/loggerdataresp"
	"github.com/fiskerinc/cloud-services/pkg/tmobile"
	
	"google.golang.org/protobuf/proto"
)

// StartConsumer runs consumer and puts events into a channel for router
func StartConsumer(topic, oldTopic string) {
	defer func() {
		if err := recover(); err != nil {
			logger.Error().Msgf("PanicConsumer %v", err)
		}
	}()

	events := make(chan *kafka.Message)
	eventsJSON := make(chan common.EventRawJSON)

	go routeEvents(events)
	go routeOldEvents(eventsJSON)

	logger.Info().Msgf("consumer intialized for topic: %v", topic)
	consumer, oldConsumer, err := services.GetKafkaConsumer()
	if err != nil {
		panic(err)
	}
	go func() {
		err = oldConsumer.ConsumeToChannelJson([]string{oldTopic}, eventsJSON)
		loggerdataresp.BadDataError(err, loggerdataresp.EofErrorCheck)
	}()

	err = consumer.ConsumeToChannel([]string{topic}, events)
	loggerdataresp.BadDataError(err, loggerdataresp.EofErrorCheck)
}

func routeEvents(events chan *kafka.Message) {
	db := services.GetDB()
	defer db.Close()

	sms := services.GetSMSClient()
	ka := services.NewKeepAwakeService()

	for {
		event := <-events
		var err error
		payload := &kafka_grpc.GRPC_AttendantPayload{}
		err = proto.Unmarshal(event.Value, payload)
		if err != nil {
			logger.Warn().Err(err).Send()
		}
		device, key := common.ParseDeviceKey(string(event.Key))
		logger.Debug().Str("id", key).Msgf("source: %s, type: %s, handler: %s", key, device, payload.GetHandler())

		switch device {
		case common.TRex:
			if payload.GetHandler() == "dtcs" {
				d := &common.ConsumerPayload{
					Handler: payload.GetHandler(),
					Data:    controllers.GRPCToDTCEntry(payload),
				}
				routeService(db, ka, sms, key, d)
				break
			}
			d, _ := common.AttendantRouteTRexPayload(payload)
			routeTRex(db, ka, key, d)
		case common.HMI:
			d, _ := common.AttendantRouteHMIPayload(payload)
			routeHMI(db, ka, key, d)
		case common.Mobile:
			d, _ := common.AttendantRouteMobilePayload(payload)
			routeMobile(db, key, d)
		case common.Service:
			if payload.GetHandler() == "sms_delivery_status_manifest" {
				d := &common.ConsumerPayload{
					Handler: payload.GetHandler(),
					Data:    tmobile.GRPCToTMessage(payload),
				}
				routeService(db, ka, sms, key, d)
				break
			}
			d, _ := common.AttendantRouteServicePayload(payload)
			routeService(db, ka, sms, key, d)
		default:
			{
				logger.Error().Str("id", key).Msgf("Unknown device: %s, type: %s, handler: %s", device, key, payload.GetHandler())

				loggerdataresp.BadDataError(ErrInvalidDevice)
			}
		}
		loggerdataresp.BadDataError(err)
	}
}

func routeOldEvents(events chan common.EventRawJSON) {
	db := services.GetDB()
	defer db.Close()

	sms := services.GetSMSClient()
	ka := services.NewKeepAwakeService()

	for {
		event := <-events
		var p common.Payload
		err := p.Unmarshal(event.Payload)
		device, key := common.ParseDeviceKey(event.Key)
		payload := &common.ConsumerPayload{
			Handler: p.Handler,
			Data:    p.Data,
		}
		logger.Debug().Str("id", key).Msgf("source: %s, type: %s, handler: %s", key, device, payload.GetHandler())
		switch device {
		case common.TRex:
			routeTRex(db, ka, key, payload)
		case common.HMI:
			routeHMI(db, ka, key, payload)
		case common.Mobile:
			routeMobile(db, key, payload)
		case common.Service:
			routeService(db, ka, sms, key, payload)
		default:
			{
				logger.Error().Str("id", key).Msgf("Unknown device: %s, type: %s, handler: %s", device, key, p.Handler)

				loggerdataresp.BadDataError(ErrInvalidDevice)
			}
		}
		loggerdataresp.BadDataError(err)
	}
}

func routeTRex(db *services.DB, ka *services.KeepAwake, id string, p common.ConsumerPayloadInterface) {
	// route TRex messages. Also update cloud/modules_go/kafka/topics.go
	var err error

	switch p.GetHandler() {
	case "car_state":
		err = handlers.UpdateCarState(db, id, p.GetData())
		loggerdataresp.BadDataError(err)
	case "car_update_status", "car_update_download", "car_update_install":
		err = handlers.CarUpdateProgressStatus(db, ka, common.TRex, id, p.GetData())
		loggerdataresp.BadDataError(err)
	case "get_filekeys":
		err = handlers.GetFileKeys(db, common.TRex, id, p.GetData())
	case "dtcs":
		err = handlers.UploadDtc(id, p.GetData())
	case "ecc_keys":
		err = handlers.GetAllEccKeys(db, id, p.GetData())
	default:
		err = kafka.ErrUnhandledMessage(common.TRex, id, p.GetHandler(), string(p.GetData()))
	}
	loggerdataresp.BadDataError(err)

}

func routeHMI(db *services.DB, ka *services.KeepAwake, id string, p common.ConsumerPayloadInterface) {
	// route HMI messages. Also update cloud/modules_go/kafka/topics.go
	var err error

	switch p.GetHandler() {
	case "update_approve":
		err = handlers.ApproveUpdate(db, id, p.GetData())
	case "car_update_status", "car_update_download", "car_update_install":
		go func() {
			err = handlers.CarUpdateProgressStatus(db, ka, common.HMI, id, p.GetData())
			loggerdataresp.BadDataError(err)
		}()
		err = handlers.CarUpdateProgressStatus(db, ka, common.HMI, id, p.GetData())
		loggerdataresp.BadDataError(err)
	case "get_filekeys":
		err = handlers.GetFileKeys(db, common.HMI, id, p.GetData())
	default:
		err = kafka.ErrUnhandledMessage(common.HMI, id, p.GetHandler(), string(p.GetData()))
	}

	loggerdataresp.BadDataError(err)
}

func routeMobile(db *services.DB, id string, p common.ConsumerPayloadInterface) {
	// route mobile messages. Also update cloud/modules_go/kafka/topics.go
	var err error

	switch p.GetHandler() {
	case "update_approve":
		err = handlers.ApproveUpdate(db, id, p.GetData())
	case "updates_get":
		err = handlers.GetUpdates(db, id, p.GetData())
	default:
		err = kafka.ErrUnhandledMessage(common.Mobile, id, p.GetHandler(), string(p.GetData()))
	}
	loggerdataresp.BadDataError(err)
}

func routeService(db *services.DB, ka *services.KeepAwake, sms sms.SMSServiceClient, id string, p common.ConsumerPayloadInterface) {
	// route cloud service messages. Also update cloud/modules_go/kafka/topics.go
	var err error

	// We should switch out these strings for constants in the kafka library
	switch p.GetHandler() {
	case "send_manifest":
		err = handlers.SendManifest(db, ka, common.Service, id, p.GetData())
	case "order_updated":
		err = handlers.OrderUpdated(db, sms, id, p.GetData())
	case "sms_delivery_status_manifest":
		err = handlers.CarSendManifestSMSWakeUpCallback(db, ka, common.Service, id, p.GetData())
	default:
		err = kafka.ErrUnhandledMessage(common.Service, id, p.GetHandler(), string(p.GetData()))
	}

	loggerdataresp.BadDataError(err)
}