scala – spray.can.Http $ConnectionException：过早连接关闭

发布时间：2020-12-16 18:16:37 所属栏目：安全来源：网络整理

导读：在我的下面测试中,我尝试模拟超时,然后发送正常请求.但是,我得到了spray.can.Http $ConnectionException：过早连接关闭(服务器似乎不支持请求流水线) class SprayCanTest extends ModuleTestKit("/SprayCanTest.conf") with FlatSpecLike with Matchers { im

在我的下面测试中,我尝试模拟超时,然后发送正常请求.但是,我得到了spray.can.Http $ConnectionException：过早连接关闭(服务器似乎不支持请求流水线)

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


}

和我的测试配置：

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

我发现在这两种情况下,“conn”对象都是一样的.
所以我想当RequestTimeoutException发生时,将conn放回池中(默认为4？),下一个案例将使用相同的conn但此时此conn保持活动状态,因此服务器会将其视为分块请求.

如果我在第二种情况下睡一觉,它就会过去.
所以我想我必须在得到RequestTimeoutException时关闭conn并确保第二种情况使用一个全新的连接,对吧？

我应该怎么做？任何配置？

谢谢

莱昂

解决方法

你不应该阻止一个Actor(你的MockServer).当它被阻止时,它无法响应任何消息.您可以将Thread.sleep和响应包装在Future中.甚至更好：使用 Akka Scheduler.确保将发送方分配给val,因为它可能会在您异步响应请求时发生更改.这应该做的伎俩：

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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!