spray.can.Http $ ConnectionException: Vorzeitige Verbindung schließen

Question

In meinem unten genannten Test habe ich versucht, ein Timeout zu simulieren und dann eine normale Anfrage zu senden. jedoch habe ich spray.can.Http $ Connection: Vorzeitige Verbindung schließen (der Server nicht erscheint Anfrage unterstützt 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")) 
     } 
    } 
    } 


} 

und meine Config für den Test:

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

Ich fand, dass In beiden Fällen ist das "conn" -Objekt dasselbe. Also ich denke, wenn RequestTimeoutException passiert, Spray setzen Sie die Conn in den Pool (standardmäßig 4?) Und der nächste Fall wird die gleiche Verbindung, aber zu diesem Zeitpunkt, diese Conn am Leben erhalten, so wird der Server es als Chunked behandeln anfordern.

Wenn ich etwas Schlaf im zweiten Fall, wird es gerade bestanden. Also ich denke, ich muss die Verbindung schließen, wenn RequestTimeoutException und sicherstellen, dass der zweite Fall eine neue neue Verbindung verwenden, nicht wahr?

Wie soll ich tun? Irgendwelche Konfigurationen?

Dank

Leon

Answer 1

Sie sollten sich nicht in einem Darsteller (Ihre MockServer) blockieren. Wenn es blockiert ist, kann es auf keine Nachrichten antworten. Sie können das Thread.sleep und die Antwort innerhalb eines Futures einpacken. Oder noch besser: Verwenden Sie die Akka Scheduler. Weisen Sie den Absender einem Wert zu, da er sich ändern kann, wenn Sie auf die Anforderung asynchron antworten. Das sollte den Trick machen:

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