<commit_before>package pl.newicom.eventstore

import akka.actor._
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, OverflowStrategy}
import eventstore.EventNumber._
import eventstore._
import eventstore.pipeline.TickGenerator.{Tick, Trigger}
import pl.newicom.dddd.messaging.event.EventStreamSubscriber.{InFlightMessagesCallback, EventReceived}
import pl.newicom.dddd.messaging.event.{EventStream, _}

class DemandController(triggerActor: ActorRef, bufferSize: Int, initialDemand: Int = 1) extends InFlightMessagesCallback {

  increaseDemand(initialDemand)

  def onChanged(messagesInFlight: Int): Unit = {
    increaseDemand(bufferSize - messagesInFlight)
  }

  private def increaseDemand(increaseValue: Int): Unit =
    for (i <- 1 to increaseValue)
      triggerActor ! Tick(null)
}

trait EventstoreSubscriber extends EventStreamSubscriber with EventstoreSerializationSupport with ActorLogging {
  this: Actor =>

  override def system = context.system

  def bufferSize: Int = 10

  implicit val actorMaterializer = ActorMaterializer()

  def subscribe(stream: EventStream, fromPosExcl: Option[Long]): InFlightMessagesCallback = {

    def eventSource: Source[EventReceived, Unit] = {
      def withMetaData(eventData: EventData): EventMessage = {
        val em = toEventMessage(eventData).get
        em.withMetaData(metaDataProvider(em))
      }
      val streamId = StreamNameResolver.streamId(stream)
      log.debug(s"Subscribing to $streamId from position $fromPosExcl (exclusive)")
      Source(
        EsConnection(system).streamPublisher(
          streamId,
          fromPosExcl.map(l => Exact(l.toInt)),
          resolveLinkTos = true
        )
      ).map {
        case EventRecord(_, number, eventData, _) =>
          EventReceived(withMetaData(eventData), number.value)
        case ResolvedEvent(EventRecord(_, _, eventData, _), linkEvent) =>
          EventReceived(withMetaData(eventData), linkEvent.number.value)
      }
    }

   def flow: Flow[Trigger, EventReceived, Unit] = Flow() { implicit b =>
      import FlowGraph.Implicits._
      val zip = b.add(ZipWith((msg: EventReceived, trigger: Trigger) => msg))

      eventSource ~> zip.in0
      (zip.in1, zip.out)
    }

    val sink = Sink.actorRef(self, onCompleteMessage = Kill)
    val triggerSource = Source.actorRef(bufferSize, OverflowStrategy.fail)

    val triggerActor = flow.toMat(sink)(Keep.both).runWith(triggerSource)

    new DemandController(triggerActor, bufferSize)
  }
<commit_msg>use 'dropNew' instead of 'fail' overflow strategy<commit_after>package pl.newicom.eventstore

import akka.actor._
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, OverflowStrategy}
import eventstore.EventNumber._
import eventstore._
import eventstore.pipeline.TickGenerator.{Tick, Trigger}
import pl.newicom.dddd.messaging.event.EventStreamSubscriber.{InFlightMessagesCallback, EventReceived}
import pl.newicom.dddd.messaging.event.{EventStream, _}

class DemandController(triggerActor: ActorRef, bufferSize: Int, initialDemand: Int = 1) extends InFlightMessagesCallback {

  increaseDemand(initialDemand)

  def onChanged(messagesInFlight: Int): Unit = {
    increaseDemand(bufferSize - messagesInFlight)
  }

  private def increaseDemand(increaseValue: Int): Unit =
    for (i <- 1 to increaseValue)
      triggerActor ! Tick(null)
}

trait EventstoreSubscriber extends EventStreamSubscriber with EventstoreSerializationSupport with ActorLogging {
  this: Actor =>

  override def system = context.system

  def bufferSize: Int = 10

  implicit val actorMaterializer = ActorMaterializer()

  def subscribe(stream: EventStream, fromPosExcl: Option[Long]): InFlightMessagesCallback = {

    def eventSource: Source[EventReceived, Unit] = {
      def withMetaData(eventData: EventData): EventMessage = {
        val em = toEventMessage(eventData).get
        em.withMetaData(metaDataProvider(em))
      }
      val streamId = StreamNameResolver.streamId(stream)
      log.debug(s"Subscribing to $streamId from position $fromPosExcl (exclusive)")
      Source(
        EsConnection(system).streamPublisher(
          streamId,
          fromPosExcl.map(l => Exact(l.toInt)),
          resolveLinkTos = true
        )
      ).map {
        case EventRecord(_, number, eventData, _) =>
          EventReceived(withMetaData(eventData), number.value)
        case ResolvedEvent(EventRecord(_, _, eventData, _), linkEvent) =>
          EventReceived(withMetaData(eventData), linkEvent.number.value)
      }
    }

   def flow: Flow[Trigger, EventReceived, Unit] = Flow() { implicit b =>
      import FlowGraph.Implicits._
      val zip = b.add(ZipWith((msg: EventReceived, trigger: Trigger) => msg))

      eventSource ~> zip.in0
      (zip.in1, zip.out)
    }

    val sink = Sink.actorRef(self, onCompleteMessage = Kill)
    val triggerSource = Source.actorRef(bufferSize, OverflowStrategy.dropNew)

    val triggerActor = flow.toMat(sink)(Keep.both).runWith(triggerSource)

    new DemandController(triggerActor, bufferSize)
  }

{'fraction_non_alphanumeric': 0.09740645416749158, 'fraction_numerical': 0.002375767174816868, 'mean_word_length': 4.006937561942517, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 17, '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': '5901233', 'n_tokens_mistral': 1523, 'n_tokens_neox': 1482, 'n_words': 318}