<commit_before>package io.gatling.http.integration

import java.io.File

import akka.actor.{ Actor, ActorRef, Props }
import akka.io.IO
import akka.pattern.ask
import akka.testkit.{ ImplicitSender, TestKit }
import akka.util.Timeout
import com.typesafe.scalalogging.slf4j.Logging
import io.gatling.core.akka.GatlingActorSystem
import io.gatling.core.config.Protocols
import io.gatling.core.controller.DataWritersInitialized
import io.gatling.core.result.writer.{ DataWriter, RunMessage }
import io.gatling.core.session.Session
import io.gatling.core.structure.ScenarioBuilder
import io.gatling.core.test.ActorSupport
import io.gatling.http.Predef._
import io.gatling.http.ahc.{ AsyncHandlerActor, HttpEngine }
import org.joda.time.DateTime
import org.specs2.execute.{ AsResult, Result }
import org.specs2.specification.Fixture
import spray.can.Http
import spray.http._

import scala.concurrent.Await
import scala.concurrent.duration._

object MockServerSupport extends Fixture[TestKit with ImplicitSender] with Logging {

  val mockHttpPort = Option(Integer.getInteger("gatling.mockHttp.port")).map(_.intValue).getOrElse(8702)
  def httpProtocol = http.baseURL(s"http://localhost:$mockHttpPort")

  var serverActor: ActorRef = _

  var requests: List[HttpRequest] = Nil

  class HttpActor(process: PartialFunction[HttpRequest, HttpResponse]) extends Actor {

    override def receive: Receive = {
      // when a new connection comes in we register ourselves as the connection handler
      case _: Http.Connected => sender ! Http.Register(self)

      case r: HttpRequest if process.isDefinedAt(r) => {
        record(r)
        sender ! process(r)
      }

      case r: HttpRequest => {
        record(r)
        logger.warn(s"Unhandled request: $r")
        sender ! HttpResponse(404)
      }
    }

    def record(request: HttpRequest) = {
      requests = request :: requests
    }
  }

  def serverMock(f: PartialFunction[Any, HttpResponse])(implicit testKit: TestKit with ImplicitSender) = {
    serverActor = GatlingActorSystem.instance.actorOf(Props(new HttpActor(f)), "mockServer")

    implicit val timeout = Timeout(4 seconds)
    val future = IO(Http)(GatlingActorSystem.instance) ? Http.Bind(serverActor, interface = "localhost", port = mockHttpPort)
    Await.result(future, Duration.Inf)
  }

  def apply[R: AsResult](config: Map[String, _])(f: TestKit with ImplicitSender => R): Result = {
    ActorSupport(config) { implicit testKit =>
      import testKit._

      try {
        HttpEngine.start()
        AsyncHandlerActor.start()

        //Initialise DataWriter with fake data.
        DataWriter.init(RunMessage("FakeSimulation", "fakesimulation1", DateTime.now, "A fake run"), Nil, self)
        expectMsgClass(classOf[DataWritersInitialized])

        f(testKit)
      } finally {
        /*
         * DataWriter, AsyncHandlerActor and HttpEngine don't need explicit shutdown - they have callbacks registered
         * with GatlingActorSystem, so they will be shutdown with the ActorSystem
         */
        if (serverActor != null) {
          IO(Http)(GatlingActorSystem.instance) ! Http.Unbind(4 second)
        }

        serverActor = null
        requests = Nil<commit_msg>Fixed mock HTTP server stop<commit_after>package io.gatling.http.integration

import java.io.File

import akka.actor.{ Actor, ActorRef, Props }
import akka.io.{ Tcp, IO }
import akka.pattern.ask
import akka.testkit.{ ImplicitSender, TestKit }
import akka.util.Timeout
import com.typesafe.scalalogging.slf4j.Logging
import io.gatling.core.akka.GatlingActorSystem
import io.gatling.core.config.Protocols
import io.gatling.core.controller.DataWritersInitialized
import io.gatling.core.result.writer.{ DataWriter, RunMessage }
import io.gatling.core.session.Session
import io.gatling.core.structure.ScenarioBuilder
import io.gatling.core.test.ActorSupport
import io.gatling.http.Predef._
import io.gatling.http.ahc.{ AsyncHandlerActor, HttpEngine }
import org.joda.time.DateTime
import org.specs2.execute.{ AsResult, Result }
import org.specs2.specification.Fixture
import spray.can.Http
import spray.http._

import scala.concurrent.Await
import scala.concurrent.duration._

object MockServerSupport extends Fixture[TestKit with ImplicitSender] with Logging {

  val mockHttpPort = Option(Integer.getInteger("gatling.mockHttp.port")).map(_.intValue).getOrElse(8702)
  def httpProtocol = http.baseURL(s"http://localhost:$mockHttpPort")

  var serverActor: ActorRef = _

  var requests: List[HttpRequest] = Nil

  class HttpActor(process: PartialFunction[HttpRequest, HttpResponse]) extends Actor {

    var bindSender: ActorRef = _
    var httpListener: ActorRef = _
    var unbindSender: ActorRef = _

    override def receive: Receive = {

      case bind: Http.Bind => {
        bindSender = sender
        IO(Http)(GatlingActorSystem.instance) ! bind
      }

      case bound: Http.Bound => {
        httpListener = sender
        bindSender ! bound
      }

      // when a new connection comes in we register ourselves as the connection handler
      case _: Http.Connected => sender ! Http.Register(self)

      case unbind: Http.Unbind => {
        unbindSender = sender
        httpListener ! unbind
      }

      case Http.Unbound => {
        unbindSender ! Http.Unbound
      }

      case r: HttpRequest if process.isDefinedAt(r) => {
        record(r)
        sender ! process(r)
      }

      case r: HttpRequest => {
        record(r)
        logger.warn(s"Unhandled request: $r")
        sender ! HttpResponse(404)
      }

    }

    def record(request: HttpRequest) = {
      requests = request :: requests
    }
  }

  def serverMock(f: PartialFunction[Any, HttpResponse])(implicit testKit: TestKit with ImplicitSender) = {
    serverActor = GatlingActorSystem.instance.actorOf(Props(new HttpActor(f)), "mockServerActor")
    implicit val timout = Timeout(4 seconds)
    val future = serverActor ? Http.Bind(serverActor, interface = "localhost", port = mockHttpPort)
    Await.result(future, Duration.Inf)
  }

  def apply[R: AsResult](config: Map[String, _])(f: TestKit with ImplicitSender => R): Result = {
    ActorSupport(config) { implicit testKit =>
      import testKit._

      try {
        HttpEngine.start()
        AsyncHandlerActor.start()

        //Initialise DataWriter with fake data.
        DataWriter.init(RunMessage("FakeSimulation", "fakesimulation1", DateTime.now, "A fake run"), Nil, self)
        expectMsgClass(classOf[DataWritersInitialized])

        f(testKit)
      } finally {
        /*
         * DataWriter, AsyncHandlerActor and HttpEngine don't need explicit shutdown - they have callbacks registered
         * with GatlingActorSystem, so they will be shutdown with the ActorSystem
         */
        if (serverActor != null) {
          implicit val timeout = Timeout(8 seconds)
          val future = serverActor ? Http.Unbind(4 second)
          Await.result(future, Duration.Inf)
        }

        serverActor = null
        requests = Nil
{'fraction_non_alphanumeric': 0.0987302040233985, 'fraction_numerical': 0.0038521900413753746, 'mean_word_length': 3.3703241895261846, 'pattern_counts': {'":': 0, '<': 3, '<?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': '7663322', 'n_tokens_mistral': 2142, 'n_tokens_neox': 2066, 'n_words': 494}