<commit_before>package com.ubirch.transformer.actor

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.routing.RoundRobinPool
import com.ubirch.avatar.config.{Config, ConfigKeys}
import com.ubirch.avatar.core.device.DeviceHistoryManager
import com.ubirch.avatar.model._
import com.ubirch.avatar.model.db.device.Device
import com.ubirch.avatar.model.rest.device.{DeviceDataRaw, DeviceType}
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.transformer.model.MessageReceiver
import com.ubirch.transformer.services.TransformerService
import com.ubirch.util.json.{Json4sUtil, MyJsonProtocol}

import scala.language.postfixOps

/**
  * Created by derMicha on 28/10/16.
  */
class TransformerPostprocessorActor extends Actor with MyJsonProtocol with ActorLogging {

  private implicit val executionContext = context.dispatcher

  val outboxManagerActor: ActorRef = context.actorOf(Props[TransformerOutboxManagerActor], ActorNames.TRANSFORMER_OUTBOX_MANAGER)

  override def receive: Receive = {

    case (deviceType: DeviceType, device: Device, drd: DeviceDataRaw) =>
      self forward(deviceType, device, drd, drd)
    case (deviceType: DeviceType, device: Device, drd: DeviceDataRaw, sdrd: DeviceDataRaw) =>
      log.debug(s"received device preprocessed raw data message: $drd with deviceKeyType: $deviceType")

      val dbDevice = Json4sUtil.any2any[db.device.Device](device)
      TransformerService.transform(
        deviceType = deviceType,
        device = dbDevice,
        drd = drd,
        sdrd = sdrd
      ) match {
        case Some(ddp) =>
          DeviceHistoryManager.store(ddp)

          Json4sUtil.any2jvalue(ddp) match {
            case Some(jval) =>

              log.debug("send processed message to sqs")

              val jvalStr = Json4sUtil.jvalue2String(jval)

              if (device.pubQueues.isDefined) {
                device.pubQueues.get.foreach { sqsQueueName =>
                  log.debug(s"send processed message for deviceId ${device.deviceId} to $sqsQueueName")
                  outboxManagerActor ! MessageReceiver(sqsQueueName, jvalStr, ConfigKeys.INTERNOUTBOX)
                }
              }
              else
                log.info(s"no pubQueues defined for deviceId: ${device.deviceId}")

              if (Config.mqttPublishProcessed) {
                log.debug(s"send processed message to mqtt for deviceId ${device.deviceId}")
                outboxManagerActor ! MessageReceiver(ddp.deviceId, jvalStr, ConfigKeys.EXTERNOUTBOX)
              }
              else
                log.info(s"do not send processed message for deviceId ${device.deviceId} to mqtt")
<commit_msg>just a hot fix 2<commit_after>package com.ubirch.transformer.actor

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.routing.RoundRobinPool
import com.ubirch.avatar.config.{Config, ConfigKeys}
import com.ubirch.avatar.core.device.DeviceHistoryManager
import com.ubirch.avatar.model._
import com.ubirch.avatar.model.db.device.Device
import com.ubirch.avatar.model.rest.device.{DeviceDataRaw, DeviceType}
import com.ubirch.avatar.util.actor.ActorNames
import com.ubirch.transformer.model.MessageReceiver
import com.ubirch.transformer.services.TransformerService
import com.ubirch.util.json.{Json4sUtil, MyJsonProtocol}

import scala.language.postfixOps

/**
  * Created by derMicha on 28/10/16.
  */
class TransformerPostprocessorActor extends Actor with MyJsonProtocol with ActorLogging {

  private implicit val executionContext = context.dispatcher

  val outboxManagerActor: ActorRef = context.actorOf(Props[TransformerOutboxManagerActor], ActorNames.TRANSFORMER_OUTBOX_MANAGER)

  override def receive: Receive = {

    case (deviceType: DeviceType, device: Device, drd: DeviceDataRaw) =>
      self forward(deviceType, device, drd, drd)
    case (deviceType: DeviceType, device: Device, drd: DeviceDataRaw, sdrd: DeviceDataRaw) =>
      log.debug(s"received device preprocessed raw data message: $drd with deviceKeyType: $deviceType")

      val dbDevice = Json4sUtil.any2any[db.device.Device](device)
      TransformerService.transform(
        deviceType = deviceType,
        device = dbDevice,
        drd = drd,
        sdrd = sdrd
      ) match {
        case Some(ddp) =>
          DeviceHistoryManager.store(ddp)

          Json4sUtil.any2jvalue(ddp) match {
            case Some(jval) =>

              log.debug("send processed message to sqs")

              val jvalStr = Json4sUtil.jvalue2String(jval)

              if (device.pubQueues.isDefined) {
                device.pubQueues.get.foreach { sqsQueueName =>
                  log.debug(s"send processed message for deviceId ${device.deviceId} to $sqsQueueName")
                  outboxManagerActor ! MessageReceiver(sqsQueueName, jvalStr, ConfigKeys.INTERNOUTBOX)
                }
              }
              else
                log.info(s"no pubQueues defined for deviceId: ${device.deviceId}")

              if (Config.mqttPublishProcessed) {
                log.debug(s"currently do not send processed message to mqtt for deviceId ${device.deviceId}")
                //@TODO just a hot fix
                //outboxManagerActor ! MessageReceiver(ddp.deviceId, jvalStr, ConfigKeys.EXTERNOUTBOX)
              }
              else
                log.info(s"do not send processed message for deviceId ${device.deviceId} to mqtt")

{'fraction_non_alphanumeric': 0.08905758192927235, 'fraction_numerical': 0.004999074245510091, 'mean_word_length': 3.004447739065975, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '143708', 'n_tokens_mistral': 1658, 'n_tokens_neox': 1598, 'n_words': 344}