<commit_before>package com.shocktrade.narratives

import java.lang.{Double => JDouble, Long => JLong}
import java.text.SimpleDateFormat

import akka.actor.Actor
import com.ldaniels528.broadway.BroadwayNarrative
import com.ldaniels528.broadway.core.actors.Actors.Implicits._
import com.ldaniels528.broadway.core.actors.Actors._
import com.ldaniels528.broadway.core.actors.FileReadingActor.{CopyText, Delimited, _}
import com.ldaniels528.broadway.core.actors.kafka.avro.KafkaAvroPublishingActor
import com.ldaniels528.broadway.core.actors.{FileReadingActor, ThrottlingActor}
import com.ldaniels528.broadway.core.resources._
import com.ldaniels528.broadway.server.ServerConfig
import com.ldaniels528.trifecta.util.StringHelper._
import com.shocktrade.helpers.ConversionHelper._
import com.shocktrade.helpers.ResourceTracker
import com.shocktrade.narratives.EodDataImportNarrative._

/**
 * EODData.com Import Narrative
 * @author <NAME> <<EMAIL>>
 */
class EodDataImportNarrative(config: ServerConfig) extends BroadwayNarrative(config, "EOD Data Import")
with KafkaConstants {

  onStart { resource =>
    // create a file reader actor to read lines from the incoming resource
    val fileReader = addActor(new FileReadingActor(config))

    // create a Kafka publishing actor
    val kafkaPublisher = addActor(new KafkaAvroPublishingActor(eodDataTopic, brokers))

    // let's throttle the messages flowing into Kafka
    val throttler = addActor(new ThrottlingActor(config, kafkaPublisher, rateLimit = 20))

    // create a EOD data transformation actor
    val eodDataToAvroActor = addActor(new EodDataToAvroActor(throttler))

    // start the processing by submitting a request to the file reader actor
    fileReader ! CopyText(resource, eodDataToAvroActor, handler = Delimited("[,]"))
  }
}

/**
 * EODData.com Import Narrative Singleton
 * @author <NAME> <<EMAIL>>
 */
object EodDataImportNarrative {

  /**
   * EODData-to-Avro Actor
   * @author <NAME> <<EMAIL>>
   */
  class EodDataToAvroActor(target: BWxActorRef) extends Actor {
    private val sdf = new SimpleDateFormat("yyyyMMdd")

    override def receive = {
      case OpeningFile(resource) =>
        ResourceTracker.start(resource)

      case ClosingFile(resource) =>
        ResourceTracker.stop(resource)

      case TextLine(resource, lineNo, line, tokens) =>
        // skip the header line
        if (lineNo != 1) {
          target ! toAvro(resource, tokens)
        }

      case message =>
        unhandled(message)
    }

    /**<commit_msg>testing throttling of Kafka publishing<commit_after>package com.shocktrade.narratives

import java.lang.{Double => JDouble, Long => JLong}
import java.text.SimpleDateFormat

import akka.actor.Actor
import com.ldaniels528.broadway.BroadwayNarrative
import com.ldaniels528.broadway.core.actors.Actors.Implicits._
import com.ldaniels528.broadway.core.actors.Actors._
import com.ldaniels528.broadway.core.actors.FileReadingActor.{CopyText, Delimited, _}
import com.ldaniels528.broadway.core.actors.kafka.avro.KafkaAvroPublishingActor
import com.ldaniels528.broadway.core.actors.kafka.avro.KafkaAvroPublishingActor._
import com.ldaniels528.broadway.core.actors.{FileReadingActor, ThrottlingActor}
import com.ldaniels528.broadway.core.resources._
import com.ldaniels528.broadway.server.ServerConfig
import com.ldaniels528.trifecta.util.StringHelper._
import com.shocktrade.helpers.ConversionHelper._
import com.shocktrade.helpers.ResourceTracker
import com.shocktrade.narratives.EodDataImportNarrative._

/**
 * EODData.com Import Narrative
 * @author <NAME> <<EMAIL>>
 */
class EodDataImportNarrative(config: ServerConfig) extends BroadwayNarrative(config, "EOD Data Import")
with KafkaConstants {

  onStart { resource =>
    // create a file reader actor to read lines from the incoming resource
    val fileReader = addActor(new FileReadingActor(config))

    // create a Kafka publishing actor
    val kafkaPublisher = addActor(new KafkaAvroPublishingActor(eodDataTopic, brokers))

    // let's throttle the messages flowing into Kafka
    val throttler = addActor(new ThrottlingActor(config, kafkaPublisher, rateLimit = 100))

    // create a EOD data transformation actor
    val eodDataToAvroActor = addActor(new EodDataToAvroActor(throttler))

    // start the processing by submitting a request to the file reader actor
    fileReader ! CopyText(resource, eodDataToAvroActor, handler = Delimited("[,]"))
  }
}

/**
 * EODData.com Import Narrative Singleton
 * @author <NAME> <<EMAIL>>
 */
object EodDataImportNarrative {

  /**
   * EODData-to-Avro Actor
   * @author <NAME> <<EMAIL>>
   */
  class EodDataToAvroActor(kafkaActor: BWxActorRef) extends Actor {
    private val sdf = new SimpleDateFormat("yyyyMMdd")

    override def receive = {
      case OpeningFile(resource) =>
        ResourceTracker.start(resource)

      case ClosingFile(resource) =>
        ResourceTracker.stop(resource)

      case TextLine(resource, lineNo, line, tokens) =>
        // skip the header line
        if (lineNo != 1) {
          kafkaActor ! PublishAvro(record = toAvro(resource, tokens))
        }

      case message =>
        unhandled(message)
    }

    /**
{'fraction_non_alphanumeric': 0.09671814671814671, 'fraction_numerical': 0.012355212355212355, 'mean_word_length': 4.217522658610272, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 35, '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': '1116938', 'n_tokens_mistral': 1713, 'n_tokens_neox': 1585, 'n_words': 353}