Akka streams: Reading multiple files
<p>I have a list of files. I want:</p>

<ol>
<li>To read from all of them as a single Source.</li>
<li>Files should be read sequentially, in-order. (no round-robin)</li>
<li>At no point should any file be required to be entirely in memory.</li>
<li>An error reading from a file should collapse the stream.</li>
</ol>

<p>It felt like this should work: (Scala, akka-streams v2.4.7)</p>

<pre><code>val sources = Seq("file1", "file2").map(new File(_)).map(f =&gt; FileIO.fromPath(f.toPath)
    .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true))
    .map(bs =&gt; bs.utf8String)
  )
val source = sources.reduce( (a, b) =&gt; Source.combine(a, b)(MergePreferred(_)) )
source.map(_ =&gt; 1).runWith(Sink.reduce[Int](_ + _)) // counting lines
</code></pre>

<p>But that results in a compile error since <code>FileIO</code> has a materialized value associated with it, and <code>Source.combine</code> doesn't support that.</p>

<p>Mapping the materialized value away makes me wonder how file-read errors get handled, but does compile:</p>

<pre><code>val sources = Seq("file1", "file2").map(new File(_)).map(f =&gt; FileIO.fromPath(f.toPath)
    .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true))
    .map(bs =&gt; bs.utf8String)
    .mapMaterializedValue(f =&gt; NotUsed.getInstance())
  )
val source = sources.reduce( (a, b) =&gt; Source.combine(a, b)(MergePreferred(_)) )
source.map(_ =&gt; 1).runWith(Sink.reduce[Int](_ + _))  // counting lines
</code></pre>

<p>But throws an IllegalArgumentException at runtime:</p>

<pre><code>java.lang.IllegalArgumentException: requirement failed: The inlets [] and outlets [MergePreferred.out] must correspond to the inlets [MergePreferred.preferred] and outlets [MergePreferred.out]
</code></pre>