<commit_before>package eventstore
import akka.actor.ActorRef
import akka.testkit.TestActorRef
import eventstore.PersistentSubscription.{ Ack, Connect, Connected, EventAppeared }
import eventstore.ReadDirection.Forward

class PersistentSubscriptionActorSpec extends AbstractSubscriptionActorSpec {
  "PersistentSubscriptionActor" should {
    "should connect to the eventstore" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
    }

    "should send an ack after message was received" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! EventAppeared(new EventRecord(streamId, EventNumber.Exact(0), new EventData("test")))
      connection.expectMsgType[Ack]
    }

    "should subscribe if last event was passed in" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      connection.expectNoMsg()
    }

    "should trigger LiveProcessingStarted and Ack" in new PersistentSubscriptionActorScope {
      val event: Event = newEvent(1)
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! eventAppeared(event)

      expectMsg(event)
      expectAck()
      expectMsg(LiveProcessingStarted)
    }

    "should send manual ack" in new ManualAckPersistentSubscriptionActorScope {
      val event: Event = newEvent(1)
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! eventAppeared(event)

      expectMsg(event)
      expectMsg(LiveProcessingStarted)
      expectNoMsg()

      actor ! PersistentSubscriptionActor.ManualAck(event.data.eventId)
      expectAck()
    }
  }

  trait PersistentSubscriptionActorScope extends AbstractScope {
    lazy val streamId = EventStream.Id(PersistentSubscriptionActor.getClass.getSimpleName + "-" + randomUuid.toString)
    def groupName: String = randomUuid.toString

    def settings: Settings = Settings.Default.copy(readBatchSize = readBatchSize, resolveLinkTos = resolveLinkTos)

    override def createActor(): ActorRef = {
      val props = PersistentSubscriptionActor.props(
        connection = connection.ref,
        client = testActor,
        streamId = streamId,
        groupName = groupName,
        settings = settings,
        credentials = None
      )
      TestActorRef(props)
    }

    def expectAck(): Unit = connection.expectMsgType[Ack]

    def newEvent(number: Int): Event = EventRecord(streamId, EventNumber.Exact(number), mock[EventData])

    def eventAppeared(event: Event) =
      EventAppeared(event)

    def expectEvent =
      expectMsgType[EventAppeared]

    def connected(eventNumber: Option[EventNumber.Exact] = None): Connected =
      Connected(randomUuid.toString, 5000, eventNumber)
  }

  trait ManualAckPersistentSubscriptionActorScope extends PersistentSubscriptionActorScope {
    override def createActor(): ActorRef = {
      val props = PersistentSubscriptionActor.props(
        connection = connection.ref,
        client = testActor,
        streamId = streamId,
        groupName = groupName,
        settings = settings,
        credentials = None,
        autoAck = false
      )
      TestActorRef(props)
    }
  }
}<commit_msg>test for Nak using Retry policy<commit_after>package eventstore
import akka.actor.ActorRef
import akka.testkit.TestActorRef
import eventstore.PersistentSubscription._
import eventstore.ReadDirection.Forward

class PersistentSubscriptionActorSpec extends AbstractSubscriptionActorSpec {
  "PersistentSubscriptionActor" should {
    "should connect to the eventstore" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
    }

    "should send an ack after message was received" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! EventAppeared(new EventRecord(streamId, EventNumber.Exact(0), new EventData("test")))
      connection.expectMsgType[Ack]
    }

    "should subscribe if last event was passed in" in new PersistentSubscriptionActorScope {
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      connection.expectNoMsg()
    }

    "should trigger LiveProcessingStarted and Ack" in new PersistentSubscriptionActorScope {
      val event: Event = newEvent(1)
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! eventAppeared(event)

      expectMsg(event)
      expectAck()
      expectMsg(LiveProcessingStarted)
    }

    "should send manual ack" in new ManualAckPersistentSubscriptionActorScope {
      val event: Event = newEvent(1)
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! eventAppeared(event)

      expectMsg(event)
      expectMsg(LiveProcessingStarted)
      expectNoMsg()

      actor ! PersistentSubscriptionActor.ManualAck(event.data.eventId)
      expectAck()
    }

    "should send manual nak" in new ManualNakPersistentSubscriptionActorScope {
      val event: Event = newEvent(1)
      connection.expectMsgType[Connect]
      actor ! connected(Some(EventNumber.Exact(0)))
      actor ! eventAppeared(event)

      expectMsg(event)
      expectMsg(LiveProcessingStarted)
      expectNoMsg()

      actor ! PersistentSubscriptionActor.ManualNak(event.data.eventId)
      expectNak()
    }
  }

  trait PersistentSubscriptionActorScope extends AbstractScope {
    lazy val streamId = EventStream.Id(PersistentSubscriptionActor.getClass.getSimpleName + "-" + randomUuid.toString)
    def groupName: String = randomUuid.toString

    def settings: Settings = Settings.Default.copy(readBatchSize = readBatchSize, resolveLinkTos = resolveLinkTos)

    override def createActor(): ActorRef = {
      val props = PersistentSubscriptionActor.props(
        connection = connection.ref,
        client = testActor,
        streamId = streamId,
        groupName = groupName,
        settings = settings,
        credentials = None
      )
      TestActorRef(props)
    }

    def expectAck(): Unit = connection.expectMsgType[Ack]

    def expectNak(): Unit = connection.expectMsgType[Nak]

    def newEvent(number: Int): Event = EventRecord(streamId, EventNumber.Exact(number), mock[EventData])

    def eventAppeared(event: Event) =
      EventAppeared(event)

    def expectEvent =
      expectMsgType[EventAppeared]

    def connected(eventNumber: Option[EventNumber.Exact] = None): Connected =
      Connected(randomUuid.toString, 5000, eventNumber)
  }

  trait ManualAckPersistentSubscriptionActorScope extends PersistentSubscriptionActorScope {
    override def createActor(): ActorRef = {
      val props = PersistentSubscriptionActor.props(
        connection = connection.ref,
        client = testActor,
        streamId = streamId,
        groupName = groupName,
        settings = settings,
        credentials = None,
        autoAck = false
      )
      TestActorRef(props)
    }
  }

  trait ManualNakPersistentSubscriptionActorScope extends PersistentSubscriptionActorScope {
    override def createActor(): ActorRef = {
      val props = PersistentSubscriptionActor.props(
        connection = connection.ref,
        client = testActor,
        streamId = streamId,
        groupName = groupName,
        settings = settings,
        credentials = None,
        autoAck = false
      )
      TestActorRef(props)
    }
  }
}
{'fraction_non_alphanumeric': 0.0785347213071551, 'fraction_numerical': 0.0031624719989458427, 'mean_word_length': 3.5098039215686274, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2142354', 'n_tokens_mistral': 2073, 'n_tokens_neox': 2064, 'n_words': 451}