spray.can.Http $ ConnectionException: connessione prematura close

Question

Nel mio test di seguito, ho provato a simulare un timeout e quindi inviare una richiesta normale. tuttavia, ho avuto spray.can.Http $ ConnectionException: connessione precoce vicino (il server non sembra supportare richiesta pipelining)

class SprayCanTest extends ModuleTestKit("/SprayCanTest.conf") with FlatSpecLike with Matchers { 

    import system.dispatcher 

    var app = Actor.noSender 

    protected override def beforeAll(): Unit = { 
    super.beforeAll() 
    app = system.actorOf(Props(new MockServer)) 
    } 

    override protected def afterAll(): Unit = { 
    system.stop(app) 
    super.afterAll() 
    } 


    "response time out" should "work" in { 
    val setup = Http.HostConnectorSetup("localhost", 9101, false) 

    connect(setup).onComplete { 
     case Success(conn) => { 
     conn ! HttpRequest(HttpMethods.GET, "/timeout") 
     } 
    } 

    expectMsgPF() { 
     case Status.Failure(t) => 
     t shouldBe a[RequestTimeoutException] 
    } 


    } 

    "normal http response" should "work" in { 

    //Thread.sleep(5000) 
    val setup = Http.HostConnectorSetup("localhost", 9101, false) 

    connect(setup).onComplete { 
     case Success(conn) => { 
     conn ! HttpRequest(HttpMethods.GET, "/hello") 
     } 
    } 

    expectMsgPF() { 
     case HttpResponse(status, entity, _, _) => 
     status should be(StatusCodes.OK) 
     entity should be(HttpEntity("Helloworld")) 
    } 
    } 

    def connect(setup: HostConnectorSetup)(implicit system: ActorSystem) = { 
    // for the actor 'asks' 
    import system.dispatcher 
    implicit val timeout: Timeout = Timeout(1 second) 
    (IO(Http) ? setup) map { 
     case Http.HostConnectorInfo(connector, _) => connector 
    } 
    } 

    class MockServer extends Actor { 
    //implicit val timeout: Timeout = 1.second 
    implicit val system = context.system 

    // Register connection service 
    IO(Http) ! Http.Bind(self, interface = "localhost", port = 9101) 

    def receive: Actor.Receive = { 
     case _: Http.Connected => sender ! Http.Register(self) 

     case HttpRequest(GET, Uri.Path("/timeout"), _, _, _) => { 
     Thread.sleep(3000) 
     sender ! HttpResponse(entity = HttpEntity("ok")) 
     } 

     case HttpRequest(GET, Uri.Path("/hello"), _, _, _) => { 
     sender ! HttpResponse(entity = HttpEntity("Helloworld")) 
     } 
    } 
    } 


} 

e la mia configurazione per il test:

spray { 
    can { 
    client { 
     response-chunk-aggregation-limit = 0 
     connecting-timeout = 1s 
     request-timeout = 1s 
    } 
    host-connector { 
     max-retries = 0 
    } 
    } 
} 

ho scoperto che in entrambi i casi, l'oggetto "conn" è lo stesso. Quindi suppongo che quando si verifica RequestTimeoutException, spray rimetta il conn a pool (di default 4?) E il caso successivo userà lo stesso conn ma in questo momento, questa conn è mantenuta attiva, quindi il server la tratterà come chunked richiesta.

Se metto un po 'di sonno nel secondo caso, sarà appena passato. Quindi suppongo di dover chiudere il conn quando ho ottenuto RequestTimeoutException e assicurarmi che il secondo caso usi una nuova connessione nuova, giusto?

Come devo fare? Qualche configurazione?

Grazie

Leon

Answer 1

Non si dovrebbe bloccare all'interno di un attore (il vostro MockServer). Quando è bloccato, non è in grado di rispondere a nessun messaggio. Puoi avvolgere Threads.leep e risposta in un futuro. O ancora meglio: usa lo Akka Scheduler. Assicurati di assegnare il mittente a un val perché potrebbe cambiare quando rispondi alla richiesta in modo asincrono. Questo dovrebbe fare il trucco:

val savedSender = sender() 
context.system.scheduler.scheduleOnce(3 seconds){ 
    savedSender ! HttpResponse(entity = HttpEntity("ok")) 
}