[WIP] Streams 2.4.18 update #2726
Conversation
|
part of #2601 |
| { | ||
| this.AssertAllStagesStopped(() => { | ||
| TargetFile(f => { | ||
| var completion = Source.From(_testByteStrings); |
There was a problem hiding this comment.
this needs to be .From(new []{ _testByteStrings.Head())
| b => Task.FromResult(FileIO.ToFile(f)), | ||
| () => Task.FromResult(new IOResult(0, new Result<NotUsed>(NotUsed.Instance))) | ||
| ), _materializer); | ||
| completion.MapMaterializedValue(); |
There was a problem hiding this comment.
this needs to be completion.MapMaterializedValue(list => list.SelectMany(x => x)); not sure about the actual type of list but flatMap means SelectMany
| ), _materializer); | ||
| completion.MapMaterializedValue(); | ||
|
|
||
| }); |
There was a problem hiding this comment.
task needs to be awaited + CheckFileContent
| onPush: () => { }, | ||
| onUpstreamFinish: () => | ||
| { | ||
| SetKeepGoing(true); |
There was a problem hiding this comment.
Move this into GotCompletionEvent
| @@ -984,7 +994,7 @@ private sealed class SubSource : SubSourceOutlet<TIn> | |||
| { | |||
| private readonly Logic _logic; | |||
| private readonly LazySink<TIn, TMat> _stage; | |||
| private bool _completed; | |||
| internal bool Completed; | |||
There was a problem hiding this comment.
move Completed into the Logic class
| @@ -984,7 +994,7 @@ private sealed class SubSource : SubSourceOutlet<TIn> | |||
| { | |||
| private readonly Logic _logic; | |||
| private readonly LazySink<TIn, TMat> _stage; | |||
| private bool _completed; | |||
| internal bool Completed; | |||
|
|
|||
| public SubSource(Logic logic, TIn firstElement) : base(logic, "LazySink") | |||
There was a problem hiding this comment.
you get the Logic as parameter and can use _logic.Completed in the two places below.
| @@ -1005,7 +1015,7 @@ public SubSource(Logic logic, TIn firstElement) : base(logic, "LazySink") | |||
| onUpstreamFinish: () => | |||
There was a problem hiding this comment.
replace with _logic.GotCompletionEvent()
|
Partially resolved, moving to a new branch. |
Scala port of akka/akka#22033, akka/akka#22657, and akka/akka#22647