0

我正在尝试编写一个简单的 Akka.NET 流。源是一个IActorRef. 水槽是一个ISubscriber. 我正在使用 TestKit 将其实现为单元测试：

[Fact]
public void AkkaStreams_ActorSourcePublisherSink_Works()
{
    using (var materializer = Sys.Materializer())
    {
        var probe = CreateTestProbe();
        var source = Source.ActorRef<HandlerErrorEvent>(10, OverflowStrategy.DropNew);
        var subscriber = new Mock<ISubscriber<HandlerErrorEvent>>();
        var sink = Sink.FromSubscriber<HandlerErrorEvent>(subscriber.Object);
        var graph = source.ToMaterialized(sink, Keep.Both);
        var (actor, publisher) = graph.Run(materializer);

        subscriber.Verify(s => s.OnSubscribe(It.IsAny<ISubscription>()));

        var evnt = new HandlerErrorEvent("", HandlerResult.NotHandled);
        actor.Tell(evnt, ActorRefs.Nobody);

        base.AwaitCondition(() =>
        {
            try
            {
                subscriber.Verify(s => s.OnNext(It.IsAny<HandlerErrorEvent>()));
                return true;
            }
            catch(MockException)
            {
                return false;
            }
        });
    }
}

对该方法的初始Verify调用OnSubscribe顺利通过，但模拟订阅者从未收到对OnNext.

我究竟做错了什么？

运行为netcoreapp2.0. 参考：

"Akka.TestKit.Xunit2" Version="1.3.2"
"Microsoft.NET.Test.Sdk" Version="15.5.0"
"Moq" Version="4.8.0-rc1"
"xunit" Version="2.3.1"
"xunit.runner.visualstudio" Version="2.3.1"
"dotnet-xunit" Version="2.3.1"
4

1 回答 1

2

您的ISubscriber<>模拟不符合Reactive Streams 规范。它指出，为了在订阅后获得任何消息，订阅者必须首先使用ISubscription.Request(long)方法传达需求。

一般来说，如果您使用 Akka.Streams 测试套件，则不需要模拟订阅。只需下载Akka.Streams.TestKit即可获得 Akka.Streams 的扩展方法。这样你就可以通过this.CreateManualSubscriberProbe<HandlerErrorEvent>();在你的TestKit类中调用来建立一个假订阅者。它包含许多可用于断言的方法。

例子：

public class ExampleTest : TestKit
{
    [Fact]
    public void Select_should_map_output()
    {
        using (var materializer = Sys.Materializer())
        {
            // create test probe for subscriptions
            var probe = this.CreateManualSubscriberProbe<int>();

            // create flow materialized as publisher
            var publisher = Source.From(new[] { 1, 2, 3 })
                .Select(i => i + 1)
                .RunWith(Sink.AsPublisher<int>(fanout: false), materializer);

            // subscribe probe and receive subscription
            publisher.Subscribe(probe);
            var subscription = probe.ExpectSubscription();

            // request number of elements to receive, here drain source utill the end
            subscription.Request(4); 

            // validate assertions
            probe.ExpectNext(2);
            probe.ExpectNext(3);
            probe.ExpectNext(4);

            // since source had finite number of 3 elements, expect it to complete
            probe.ExpectComplete();
        }
    }
}
于 2018-02-04T09:45:11.590 回答