0

我有一个akka-gRPC服务双向流，我正在单元测试中对其进行测试。该服务有用途akka-stream，我使用TestSink.probe来测试回复消息。我收到了来自服务的消息，但是有一个与超时相关的错误，我无法弄清楚是什么原因。这是测试：

object GreeterServiceConf {
  // important to enable HTTP/2 in server ActorSystem's config
  val configServer = ConfigFactory.parseString("akka.http.server.preview.enable-http2 = on")
    .withFallback(ConfigFactory.defaultApplication())

  val configString2 =
    """
      |akka.grpc.client {
      |  "helloworld.GreeterService" {
      |    host = 127.0.0.1
      |    port = 8080
      |  }
      |}
      |""".stripMargin
  val configClient = ConfigFactory.parseString(configString2)
}

class GreeterServiceImplSpec extends TestKit(ActorSystem("GreeterServiceImplSpec", ConfigFactory.load(GreeterServiceConf.configServer)))
  with AnyWordSpecLike
  with BeforeAndAfterAll
  with Matchers
  with ScalaFutures {

  implicit val patience: PatienceConfig = PatienceConfig(scaled(5.seconds), scaled(100.millis))

  // val testKit = ActorTestKit(conf)
  val serverSystem: ActorSystem = system
  val bound = new GreeterServer(serverSystem).run()

  // make sure server is bound before using client
  bound.futureValue

  implicit val clientSystem: ActorSystem = ActorSystem("GreeterClient", ConfigFactory.load(GreeterServiceConf.configClient))

  val client = GreeterServiceClient(
    GrpcClientSettings
      .fromConfig("helloworld.GreeterService")
      .withTls(false)
  )

  override def afterAll: Unit = {
    TestKit.shutdownActorSystem(system)
    TestKit.shutdownActorSystem(clientSystem)
  }

  "GreeterService" should {
    "reply to multiple requests" in {
      import GreeterServiceData._

      val names = List("John", "Michael", "Simone")
      val expectedReply: immutable.Seq[HelloReply] = names.map { name =>
        HelloReply(s"Hello, $name -> ${mapHelloReply.getOrElse(name, "this person does not exist =(")}")
      }

      val requestStream: Source[HelloRequest, NotUsed] = Source(names).map(name => HelloRequest(name))
      val responseStream: Source[HelloReply, NotUsed] = client.sayHelloToAll(requestStream)
      val sink = TestSink.probe[HelloReply]
      val replyStream = responseStream.runWith(sink)
      replyStream
        .requestNext(HelloReply(s"Hello, John -> I killed Java"))
        .requestNext(HelloReply(s"Hello, Michael -> We are the Jacksons 5"))
        .requestNext(HelloReply(s"Hello, Simone -> I have found a job to work with Scala =)")) // THIS IS THE LINE 122 ON THE ERROR
        // .request(3)
        // .expectNextUnorderedN(expectedReply) // I also tested this but it did not work
        .expectComplete()
    }
  }
}

错误是：

断言失败：在等待 OnComplete 时，在 expectMsg 期间超时（3 秒） java.lang.AssertionError：断言失败：在 akka 等待 OnComplete 时，在 expectMsg 期间超时（3 秒） .testkit.TestKitBase.expectMsg_internal(TestKit.scala:459) 在 akka.testkit.TestKitBase.expectMsg(TestKit.scala:436) 在 akka.testkit.TestKitBase.expectMsg$(TestKit.scala:436) 在 akka.testkit.TestKit .expectMsg(TestKit.scala:969) at akka.stream.testkit.TestSubscriber$ManualProbe.expectComplete(StreamTestKit.scala:479) at com.example.helloworld.GreeterServiceImplSpec.$anonfun$new$5(GreeterServiceImplSpec.scala:121)

4

1 回答 1

0

我让它基于项目akka-grpc-quickstart-scala.g8 工作。我正在执行runForeach以运行图表并Sink在响应流上实现。然后，当响应完成后，我assert在Future[Done].

    "reply to multiple requests" in {
      import GreeterServiceData._
      import system.dispatcher

      val names = List("John", "Martin", "Michael", "UnknownPerson")
      val expectedReplySeq: immutable.Seq[HelloReply] = names.map { name =>
        HelloReply(s"Hello, $name -> ${mapHelloReply.getOrElse(name, "this person does not exist =(")}")
      }
      // println(s"expectedReplySeq: ${expectedReplySeq.foreach(println)}")

      val requestStream: Source[HelloRequest, NotUsed] = Source(names).map(name => HelloRequest(name))
      val responseStream: Source[HelloReply, NotUsed] = client.sayHelloToAll(requestStream)

      val done: Future[Done] = responseStream.runForeach { reply: HelloReply =>
        // println(s"got streaming reply: ${reply.message}")
        assert(expectedReplySeq.contains(reply))
      }
      // OR USING Sink.foreach[HelloReply])(Keep.right)
      val sinkHelloReply = Sink.foreach[HelloReply] { e =>
        println(s"element: $e")
        assert(expectedReplySeq.contains(e))
      }
      responseStream.toMat(sinkHelloReply)(Keep.right).run().onComplete {
        case Success(value) => println(s"done")
        case Failure(exception) => println(s"exception $exception")
      }
    }

只是为了保持对整个代码的引用，GreeterServiceImplSpec 类在这里。

于 2021-01-29T09:53:39.773 回答