<commit_before>package almhirt.streaming

import akka.actor._
import akka.stream.actor._

/** This thing is unbound. Handle with care! */
trait UnboundedSink[T] {
  def publish(element: T)
}

object UnboundedSink {
  def apply[T](unboundedSink: ActorRef): UnboundedSink[T] with almhirt.common.Stoppable = {
    new UnboundedSink[T] with almhirt.common.Stoppable {
      override def publish(element: T) {
        unboundedSink ! InternalUnboundedSinkMessages.Publish(element)
      }
      override def stop() {
        unboundedSink ! InternalUnboundedSinkMessages.Stop
      }
    }
  }

  def props[T]: Props = Props(new UnboundedSinkImpl[T])
}

private[almhirt] object InternalUnboundedSinkMessages {
  case object Stop
  case class Publish(element: Any)
}

private[almhirt] class UnboundedSinkImpl[T]() extends ActorPublisher[T] with ActorLogging {
  private def deliver(queue: Vector[T]): Vector[T] = {
    if (isActive) {
      val toPublish = queue.take(this.totalDemand.toInt)
      val rest = queue.drop(toPublish.size)
      toPublish.foreach(onNext)
      rest
    } else {
      queue
    }
  }

  def receiveRunning(queue: Vector[T]): Receive = {
    case InternalUnboundedSinkMessages.Publish(element) =>
      context.become(receiveRunning(deliver(queue :+ element.asInstanceOf[T])))

    case ActorPublisherMessage.Request(amount) =>
      context.become(receiveRunning(deliver(queue)))

    case ActorPublisherMessage.Cancel =>
      if (!queue.isEmpty && log.isWarningEnabled)
        log.warning(s"${queue.size} element haven't been published on cancel.")
      ()

    case InternalUnboundedSinkMessages.Stop =>
      if (!queue.isEmpty && log.isWarningEnabled)
        log.warning(s"${queue.size} element haven't been published on stop.")

      this.onComplete()
  }

  override def receive: Receive = receiveRunning(Vector.empty)

}<commit_msg>changed log entry<commit_after>package almhirt.streaming

import akka.actor._
import akka.stream.actor._

/** This thing is unbound. Handle with care! */
trait UnboundedSink[T] {
  def publish(element: T)
}

object UnboundedSink {
  def apply[T](unboundedSink: ActorRef): UnboundedSink[T] with almhirt.common.Stoppable = {
    new UnboundedSink[T] with almhirt.common.Stoppable {
      override def publish(element: T) {
        unboundedSink ! InternalUnboundedSinkMessages.Publish(element)
      }
      override def stop() {
        unboundedSink ! InternalUnboundedSinkMessages.Stop
      }
    }
  }

  def props[T]: Props = Props(new UnboundedSinkImpl[T])
}

private[almhirt] object InternalUnboundedSinkMessages {
  case object Stop
  case class Publish(element: Any)
}

private[almhirt] class UnboundedSinkImpl[T]() extends ActorPublisher[T] with ActorLogging {
  private def deliver(queue: Vector[T]): Vector[T] = {
    if (isActive) {
      val toPublish = queue.take(this.totalDemand.toInt)
      val rest = queue.drop(toPublish.size)
      toPublish.foreach(onNext)
      rest
    } else {
      queue
    }
  }

  def receiveRunning(queue: Vector[T]): Receive = {
    case InternalUnboundedSinkMessages.Publish(element) =>
      context.become(receiveRunning(deliver(queue :+ element.asInstanceOf[T])))

    case ActorPublisherMessage.Request(amount) =>
      context.become(receiveRunning(deliver(queue)))

    case ActorPublisherMessage.Cancel =>
      if (!queue.isEmpty && log.isWarningEnabled)
        log.warning(s"${queue.size} element(s) left on cancel.")
      ()

    case InternalUnboundedSinkMessages.Stop =>
      if (!queue.isEmpty && log.isWarningEnabled)
        log.warning(s"${queue.size} element(s) left on stop.")

      this.onComplete()
  }

  override def receive: Receive = receiveRunning(Vector.empty)

}
{'fraction_non_alphanumeric': 0.11105121293800539, 'fraction_numerical': 0.0, 'mean_word_length': 3.691529709228824, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '3227604', 'n_tokens_mistral': 1211, 'n_tokens_neox': 1157, 'n_words': 240}