universal-data-ingest/src/udi/dispatcher/dispatcher.go

package dispatcher

import "log"
import "udi/mqtt"
import "udi/config"
import "udi/handlers/handler"
import "udi/handlers/ttn"


var handlerMap map[string]handler.Handler = make(map[string]handler.Handler)

func InitDispatcher() {
  log.Printf("Initializing dispatcher")
  for _, handlerEntry := range config.Config.Handlers {
    log.Printf("Trying %s", handlerEntry.Name)
    switch handlerEntry.Name {
    case "TTN":
      handlerMap[handlerEntry.Name] = ttn.NewTTNHandler()
      log.Printf("TTN initialized")
    default:
      log.Fatalf("Handler %s not found", handlerEntry.Name)
    }
  }
}

func InputDispatcher() {
  for {
    select {
    case message := <- mqtt.InputChannel:
      log.Printf("Message arrived in inputDispatcher, topic: %s, payload: %s\n", message.Topic, message.Payload)

      for _, mapping := range config.Config.TopicMappings {
        log.Printf("Testing %s -> %s", mapping.Topics, mapping.Handler)
        for _, subscribedTopic := range mapping.Topics {
          log.Printf("Testing %s in %s", message.Topic, subscribedTopic)
          if mqtt.TopicMatchesSubscription(message.Topic, subscribedTopic) {
            log.Printf("Handle message in handler %s", mapping.Handler)
            handler := handlerMap[mapping.Handler]
            handler.Handle(message.Topic, string(message.Payload))
          } else {
            log.Printf("no match")
          }
        }
      }
    }
  }
}