<commit_before>package io.amient.affinity.core.actor

import java.util.concurrent.Executors

import akka.event.Logging
import io.amient.affinity.core.Record
import io.amient.affinity.core.serde.Serde
import io.amient.affinity.core.storage.EventTime
import io.amient.affinity.kafka.ManagedKafkaConsumer

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.reflect.ClassTag

trait GatewayStream extends Gateway {

  private val log = Logging.getLogger(context.system, this)

  private val config = context.system.settings.config

  private val inputStreamProcessors = new mutable.ListBuffer[Runnable]

  private val inputStreamExecutor = Executors.newCachedThreadPool()

  @volatile private var running = true

  def stream[K: ClassTag, V: ClassTag](identifier: String)(processor: Record[K, V] => Unit): Unit = {
    val streamConfig = config.getConfig("affinity.node.gateway.stream.r10-xml-input")
    inputStreamProcessors += new Runnable {
      val inputTopics = streamConfig.getStringList("topics").toSet
      val minTimestamp = streamConfig.getLong("min.timestamp")
      val keySerde = Serde.of[K](config)
      val valSerde = Serde.of[V](config)
      override def run(): Unit = {
        val consumer: ManagedKafkaConsumer = Class.forName("io.amient.affinity.kafka.ManagedKafkaConsumerImpl")
          .asSubclass(classOf[ManagedKafkaConsumer]).newInstance()
        try {
          consumer.subscribe(streamConfig.getConfig("consumer"), inputTopics)
          log.info(s"Starting input stream processor: $identifier, min.timestamp: ${EventTime.localTime(minTimestamp)}, topics: $inputTopics")
          while (running) {
            for (record: Record[Array[Byte], Array[Byte]] <- consumer.fetch(minTimestamp)) {
              val key: K = keySerde.fromBytes(record.key)
              val value: V = valSerde.fromBytes(record.value)
              processor(new Record(key, value, record.timestamp))
            }
          }
        } finally {
          consumer.close()
        }
      }
    }
  }

  override def preStart(): Unit = {
    //FIXME this code is prone to hangs - there is no way of controlling an actor's preStart activity
    //FIXME also this code assumes the key is avro serialized
//    maybeStartMigration()
    super.preStart()
  }

  override def postStop(): Unit = {
    running = false
    super.postStop()
  }

  override def onClusterStatus(suspended: Boolean) = {
    super.onClusterStatus(suspended)
    if (!suspended && running) {
      maybeStartInputStreams()<commit_msg>gateway stream minor bug fix<commit_after>package io.amient.affinity.core.actor

import java.util.concurrent.Executors

import akka.event.Logging
import io.amient.affinity.core.Record
import io.amient.affinity.core.serde.Serde
import io.amient.affinity.core.storage.EventTime
import io.amient.affinity.kafka.ManagedKafkaConsumer

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.reflect.ClassTag

trait GatewayStream extends Gateway {

  private val log = Logging.getLogger(context.system, this)

  private val config = context.system.settings.config

  private val inputStreamProcessors = new mutable.ListBuffer[Runnable]

  private val inputStreamExecutor = Executors.newCachedThreadPool()

  @volatile private var running = true

  def stream[K: ClassTag, V: ClassTag](identifier: String)(processor: Record[K, V] => Unit): Unit = {
    val streamConfig = config.getConfig(s"affinity.node.gateway.stream.$identifier")
    inputStreamProcessors += new Runnable {
      val inputTopics = streamConfig.getStringList("topics").toSet
      val minTimestamp = streamConfig.getLong("min.timestamp")
      val keySerde = Serde.of[K](config)
      val valSerde = Serde.of[V](config)
      override def run(): Unit = {
        val consumer: ManagedKafkaConsumer = Class.forName("io.amient.affinity.kafka.ManagedKafkaConsumerImpl")
          .asSubclass(classOf[ManagedKafkaConsumer]).newInstance()
        try {
          consumer.initialize(streamConfig.getConfig("consumer"), inputTopics)
          log.info(s"Starting input stream processor: $identifier, min.timestamp: ${EventTime.localTime(minTimestamp)}, topics: $inputTopics")
          while (running) {
            for (record: Record[Array[Byte], Array[Byte]] <- consumer.fetch(minTimestamp)) {
              val key: K = keySerde.fromBytes(record.key)
              val value: V = valSerde.fromBytes(record.value)
              processor(new Record(key, value, record.timestamp))
            }
          }
        } finally {
          consumer.close()
        }
      }
    }
  }

  override def preStart(): Unit = {
    //FIXME this code is prone to hangs - there is no way of controlling an actor's preStart activity
    //FIXME also this code assumes the key is avro serialized
//    maybeStartMigration()
    super.preStart()
  }

  override def postStop(): Unit = {
    running = false
    super.postStop()
  }

  override def onClusterStatus(suspended: Boolean) = {
    super.onClusterStatus(suspended)
    if (!suspended && running) {
      maybeStartInputStreams()
{'fraction_non_alphanumeric': 0.10346176413064737, 'fraction_numerical': 0.00039115978877371407, 'mean_word_length': 3.435385949696444, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, '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': '6988378', 'n_tokens_mistral': 1482, 'n_tokens_neox': 1436, 'n_words': 335}