在 GraphDSL 中换行

Question

我是 akka 流的 GraphDSL 的新手。我正在为我们拥有的测试框架做 DSL，到目前为止一切顺利，但我面临的问题是当人们使用 DSL

时我无法换行

这里有一个例子：

Given("A url='http://localhost:8080/hello_dls_world'") ~> When(s"I make a request") ~> AndThen("I expect to receive 200")

如您所见，这条线很大，而且可能更糟。

我使用生成器创建了一个解决方案，但我认为它不是很好

object TestDSL extends MainDSL with App {

  import GraphDSL.Implicits._

  val runnableGraph = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>

    val in = newLine(builder).in
    val out = newLine(builder).out

    Given("This_is_a_hello_world") ~> When(s"I change character _") ~> in

    out ~> AndThen("I expect to receive 200") ~> Then("I expect to receive 200")

    ClosedShape
  })

  runnableGraph.run()
}

这是我的 DSL 实现

   class MainDSL {

  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  def Given(sentence: String) = Source.single(sentence.toUpperCase)

  def When(sentence: String) = Flow[Any].map(x => x.asInstanceOf[String].replace("_", " "))

  def Then(sentence: String) = Sink.foreach[Any](x => println(s"######## $x"))

  def AndThen(sentence: String) = Flow[Any].map(x => x)

  def newLine(builder: Builder[NotUsed]) = builder.add(Flow[Any].map(x => x))

}

这里有人可以给我一个更好的解决方案吗？

当我运行它时，我收到此错误

Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: The inlets [] and outlets [] must correspond to the inlets [Map.in] and outlets [Map.out]
    at scala.Predef$.require(Predef.scala:224)
    at akka.stream.Shape.requireSamePortsAs(Shape.scala:168)
    at akka.stream.impl.StreamLayout$CompositeModule.replaceShape(StreamLayout.scala:427)
    at akka.stream.scaladsl.GraphApply$class.create(GraphApply.scala:19)
    at akka.stream.scaladsl.GraphDSL$.create(Graph.scala:993)
    at stream.dsl.TestDSL$.delayedEndpoint$stream$dsl$TestDSL(TestDSL.scala:22)
    at stream.dsl.TestDSL$delayedInit$body.apply(TestDSL.scala:18)
    at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
    at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
    at scala.App$$anonfun$main.apply(App.scala:76)
    at scala.App$$anonfun$main.apply(App.scala:76)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
    at scala.App$class.main(App.scala:76)
    at stream.dsl.TestDSL$.main(TestDSL.scala:18)
    at stream.dsl.TestDSL.main(TestDSL.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)

此致

Answer 1

您应该能够通过指定流形状的 in 和 out 端口来拆分，例如

val when = builder.add(When(s"I change character _"))

Given("This_is_a_hello_world") ~> when.in
when.out ~> AndThen("I expect to receive 200")

在 GraphDSL 中换行

Make new line in GraphDSL

scala

akka

akka-stream