Akka.Net 文件 IO - 仅处理文件的前 n 个字节答案

【问题标题】：Akka.Net file IO - process only the first n bytes of a fileAkka.Net 文件 IO - 仅处理文件的前 n 个字节
【发布时间】：2018-05-17 14:06:06
【问题描述】：

我有一个非常简单的流来处理文件 - 只是一个 FileIO.FromFile 源、一个传递流和一个 Last 接收器：

    var source = FileIO.FromFile(new FileInfo(fileName));

    var flow = Flow.FromFunction<ByteString, ByteString>(x =>
    {
       Log.Info(x.Count.ToString());
       return x;
    });

    var sink = Sink.Last<ByteString>();

    var runnable = source.Via(flow).ToMaterialized(sink, Keep.Right);
    var result = runnable.Run(Context.Materializer()).Result;

流按预期运行：记录器发送字节串大小的垃圾邮件，直到文件源完全耗尽。

现在，我将接收器更改为使用 First 而不是 Last

    var source = FileIO.FromFile(new FileInfo(fileName));

    var flow = Flow.FromFunction<ByteString, ByteString>(x =>
    {
       Log.Info(x.Count.ToString());
       return x;
    });

    var sink = Sink.First<ByteString>();

    var runnable = source.Via(flow).ToMaterialized(sink, Keep.Right);
    var result = runnable.Run(Context.Materializer()).Result;

“First”的文档声明“收到一个元素后取消”，我认为这意味着接收器发出取消流的信号，这将关闭源。但是当这个流运行时，会发生两件事。

1) 我收到以下调试日志消息

[DEBUG][17/05/2018 13:55:16][Thread 0004][akka://Demo/user/DATReader/StreamSupervisor-0/Flow-0-1-fileSource] Unhandled message from akka://Demo/user/DATReader/StreamSupervisor-0/Flow-0-0-unknown-operation : Akka.Streams.Actors.Cancel

和

2) 文件被锁定，因此任何进一步的读取尝试都会失败，并出现访问被拒绝异常。

我也尝试在源上使用Take(1)，但看到了相同的效果。

我的问题是：我如何只读取文件的前 n 个字节并优雅地关闭流，以便释放任何锁（由 FileIO.FromFile 获得）？

【问题讨论】：

标签： akka-stream akka.net

【解决方案1】：

就我而言，没有这样的错误。尝试使用 sharedKillSwitch 来控制完成。关于Dynamic stream handling 的文档在这里

            var sharedKillSwitch = KillSwitches.Shared("my-kill-switch");
            var source = FileIO.FromFile(new FileInfo(fileName), chunkSize: 1024)
                    .Via(sharedKillSwitch.Flow<ByteString>());

            var sink = Sink.First<ByteString>();

            var runnable = source.Select(x => {                 
                Log.Info(x.Count.ToString());
                sharedKillSwitch.Shutdown();
                return x;
            })
            .ToMaterialized(sink, Keep.Right);

            var result = runnable.Run(materializer).Result;

【讨论】：