java - 使用 Java 在 akka 测试套件中测试流

Question

https://doc.akka.io/docs/akka/current/stream/stream-testkit.html

我正在使用 Java 使用 akka 谁能告诉我系统是如何在代码中初始化的

final Sink<Integer, CompletionStage<Integer>> sinkUnderTest =
Flow.of(Integer.class)
    .map(i -> i * 2)
    .toMat(Sink.fold(0, (agg, next) -> agg + next), Keep.right());

final CompletionStage<Integer> future =
Source.from(Arrays.asList(1, 2, 3, 4)).runWith(sinkUnderTest, system);
final Integer result = future.toCompletableFuture().get(3, TimeUnit.SECONDS);
assert (result == 20);

static ActorSystem system =ActorSystem.create() 不工作

Source.from(Arrays.asList(1, 2, 3, 4)).runWith(sinkUnderTest, system);

Answer 1

编辑:

当您使用 akka 2.5 和 2.12 api 时，您将必须按照匹配文档的快速入门部分中的描述创建一个物化器，查看 here .

private ActorSystem system;
private Materializer materializer;

@BeforeEach
public void setup() {
    system = ActorSystem.create("StreamTestKitDocTest");
    materializer = ActorMaterializer.create(system);
}

// ...

@Test
public void test() throws InterruptedException, ExecutionException, TimeoutException {
    // ...
    Source.from(Arrays.asList(1, 2, 3, 4)).runWith(sinkUnderTest, materializer);
    // ...
}

---

正如您所说，它甚至无法为您编译，我假设您的依赖项或导入存在问题。一个常见的错误是您不小心导入了 scala 版本而不是 java dsl。

这里是我用来验证其工作的依赖项和代码(使用 Java 1.8 测试):

依赖关系:

<dependencies>
    <dependency>
      <groupId>com.typesafe.akka</groupId>
      <artifactId>akka-stream-testkit_2.13</artifactId>
      <version>2.6.0</version>
      <scope>test</scope>
    </dependency>
     <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

JUnit 测试用例:

import java.util.Arrays;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import akka.actor.ActorSystem;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Keep;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

public class AkkaTest {

    private ActorSystem system;

    @Before
    public void setup() {
        system = ActorSystem.create("StreamTestKitDocTest");
    }

    @After
    public void shutdown() {
        system.terminate();
    }

    @Test
    public void test() throws InterruptedException, ExecutionException, TimeoutException {

        final Sink<Integer, CompletionStage<Integer>> sinkUnderTest =
        Flow.of(Integer.class)
            .map(i -> i * 2)
            .toMat(Sink.fold(0, (agg, next) -> agg + next), Keep.right());

        final CompletionStage<Integer> future =
        Source.from(Arrays.asList(1, 2, 3, 4)).runWith(sinkUnderTest, system);
        final Integer result = future.toCompletableFuture().get(3, TimeUnit.SECONDS);

        Assert.assertEquals(20, result.intValue());
    }
}