spray.can.Http $ ConnectionException: Przedwczesne zamknięcie połączenia

Question

W poniższym teście próbowałem zasymulować limit czasu, a następnie wysłać normalne żądanie. Jednak mam spray.can.Http $ ConnectionException: Przedwczesny ścisły związek (serwer nie wydaje się, aby obsługiwać żądania potoku)

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")) 
     } 
    } 
    } 


} 

i Moje config do badań:

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

Okazało się, że w obu przypadkach obiekt "conn" jest taki sam. Zgaduję więc, że gdy wystąpi wyjątek RequestTimeoutException, spryskaj ponownie połączenie z pulą (domyślnie 4?), A następny przypadek użyje tego samego połączenia, ale w tej chwili to połączenie jest utrzymywane przy życiu, więc serwer potraktuje go jako porwany żądanie.

Jeśli położę trochę snu w drugim przypadku, po prostu minie. Więc domyślam się, że muszę zamknąć conn kiedy dostałem RequestTimeoutException i upewnij się, że druga sprawa używa świeżego nowego połączenia, prawda?

Co mam zrobić? Dowolne konfiguracje?

Dzięki

Leon

Answer 1

Nie należy blokować wewnątrz aktorem (swojej MockServer). Kiedy jest zablokowany, nie może odpowiadać na żadne wiadomości. Możesz owinąć Thread.sleep i odpowiedź wewnątrz przyszłości. Lub jeszcze lepiej: skorzystaj z Akka Scheduler. Pamiętaj, aby przypisać nadawcę do wartości val, ponieważ może się zmienić, gdy odpowiesz na to żądanie asynchronicznie. To powinno załatwić:

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