<commit_before>package com.ubirch.avatar.core.kafka


import akka.actor.ActorSystem
import akka.event.slf4j.Logger
import akka.http.scaladsl.HttpExt
import akka.stream.Materializer
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.core.kafka.util.{EndOfLifeUpdate, InvalidDataException, UnexpectedException}
import com.ubirch.avatar.model.db.device.Device
import com.ubirch.util.json.Json4sUtil
import com.ubirch.util.mongo.connection.MongoUtil
import org.json4s.JObject
import org.json4s.JsonAST.JField
import org.json4s.JsonDSL._

import scala.concurrent.{ExecutionContext, Future}

class EndOfLifeConsumer(actorSystem: ActorSystem)(implicit mat: Materializer, httpExt: HttpExt, mongo: MongoUtil)
  extends KafkaConsumer(Set(Config.kafkaEndOfLifeTopic), Config.kafkaEndOfLifeGroup, actorSystem) {

  private val log = Logger.apply(this.getClass.getName)

  protected def handleMessage(message: String)(implicit ec: ExecutionContext): Future[Unit] = {
    log.info("proceed EndOfLifeUpdate from kafka.")

    Json4sUtil.string2JValue(message) match {
      case Some(dataJval) =>
        dataJval.extractOpt[EndOfLifeUpdate] match {
          case Some(eol) => updateEOLConfig(eol)
          case None => Future.failed(InvalidDataException(s"can't parse message as EndOfLifeUpdate: $dataJval"))
        }
      case None => Future.failed(InvalidDataException(s"message is not json format. $message"))
    }
  }

  /**
    * This method parses the configuration of a device, updates the EOL flag if necessary and
    * updates the device accordingly in the database.
    */
  def updateEOLConfig(eol: EndOfLifeUpdate)(implicit ec: ExecutionContext): Future[Unit] = {

    DeviceManager.infoByHwId(eol.hwDeviceId).map {

      case None => throw UnexpectedException(s"couldn't find device to execute EOL update $eol.")

      case Some(device: Device) if device.deviceConfig.isEmpty =>
        throw InvalidDataException(s"EOL update not possible as device config is empty for device $device")

      case Some(device: Device) =>

        val configOpt = retrieveConfigWithoutEOLifUpdateNeeded(eol, device)

        if (configOpt.isEmpty) {
          log.info(s"EOL update not necessary as config already contains 'EOL=${eol.eolReached}'")
        } else {

          val updatedDevice = addEOLtoDeviceConfig(eol, device, configOpt)
          DeviceManager.update(updatedDevice).map {
            case Some(device) =>
              log.info(s"successfully updated EOL for device with hwDeviceId ${device.hwDeviceId}")
            case None =>
              throw UnexpectedException(s"EOL update went wrong when storing updated device $updatedDevice")
          }
        }
    }
  }<commit_msg>fixing Exception type<commit_after>package com.ubirch.avatar.core.kafka


import akka.actor.ActorSystem
import akka.event.slf4j.Logger
import akka.http.scaladsl.HttpExt
import akka.stream.Materializer
import com.ubirch.avatar.config.Config
import com.ubirch.avatar.core.device.DeviceManager
import com.ubirch.avatar.core.kafka.util.{EndOfLifeUpdate, InvalidDataException, UnexpectedException}
import com.ubirch.avatar.model.db.device.Device
import com.ubirch.util.json.Json4sUtil
import com.ubirch.util.mongo.connection.MongoUtil
import org.json4s.JObject
import org.json4s.JsonAST.JField
import org.json4s.JsonDSL._

import scala.concurrent.{ExecutionContext, Future}

class EndOfLifeConsumer(actorSystem: ActorSystem)(implicit mat: Materializer, httpExt: HttpExt, mongo: MongoUtil)
  extends KafkaConsumer(Set(Config.kafkaEndOfLifeTopic), Config.kafkaEndOfLifeGroup, actorSystem) {

  private val log = Logger.apply(this.getClass.getName)

  protected def handleMessage(message: String)(implicit ec: ExecutionContext): Future[Unit] = {
    log.info("proceed EndOfLifeUpdate from kafka.")

    Json4sUtil.string2JValue(message) match {
      case Some(dataJval) =>
        dataJval.extractOpt[EndOfLifeUpdate] match {
          case Some(eol) => updateEOLConfig(eol)
          case None => Future.failed(InvalidDataException(s"can't parse message as EndOfLifeUpdate: $dataJval"))
        }
      case None => Future.failed(InvalidDataException(s"message is not json format. $message"))
    }
  }

  /**
    * This method parses the configuration of a device, updates the EOL flag if necessary and
    * updates the device accordingly in the database.
    */
  def updateEOLConfig(eol: EndOfLifeUpdate)(implicit ec: ExecutionContext): Future[Unit] = {

    DeviceManager.infoByHwId(eol.hwDeviceId).map {

      case None => throw InvalidDataException(s"couldn't find device to execute EOL update $eol.")

      case Some(device: Device) if device.deviceConfig.isEmpty =>
        throw InvalidDataException(s"EOL update not possible as device config is empty for device $device")

      case Some(device: Device) =>

        val configOpt = retrieveConfigWithoutEOLifUpdateNeeded(eol, device)

        if (configOpt.isEmpty) {
          log.info(s"EOL update not necessary as config already contains 'EOL=${eol.eolReached}'")
        } else {

          val updatedDevice = addEOLtoDeviceConfig(eol, device, configOpt)
          DeviceManager.update(updatedDevice).map {
            case Some(device) =>
              log.info(s"successfully updated EOL for device with hwDeviceId ${device.hwDeviceId}")
            case None =>
              throw UnexpectedException(s"EOL update went wrong when storing updated device $updatedDevice")
          }
        }
    }
  }
{'fraction_non_alphanumeric': 0.09335986973041432, 'fraction_numerical': 0.0025330197213678306, 'mean_word_length': 3.8704845814977973, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, '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': '3229856', 'n_tokens_mistral': 1658, 'n_tokens_neox': 1630, 'n_words': 384}