MergePrioritized graph stage #3113
Conversation
| Math.Round(ones / twos).Should().Be(2); | ||
| } | ||
|
|
||
| private RunnableGraph<T> ThreeSourceMerge<T>(Source<T, NotUsed> source1, Source<T, NotUsed> source2, |
There was a problem hiding this comment.
@Horusiath @Silv3rcircl3 I'm not sure that I've converted it properly. The original source looks like
private def threeSourceMerge[T](source1: Source[T, NotUsed], source2: Source[T, NotUsed], source3: Source[T, NotUsed], priorities: Seq[Int], probe: ManualProbe[T]) = {
RunnableGraph.fromGraph(GraphDSL.create(source1, source2, source3)((_, _, _)) { implicit b ⇒ (s1, s2, s3) ⇒
val merge = b.add(MergePrioritized[T](priorities))
s1.out ~> merge.in(0)
s2.out ~> merge.in(1)
s3.out ~> merge.in(2)
merge.out ~> Sink.fromSubscriber(probe)
ClosedShape
})
}
| builder.From(s2.Outlet).To(merge.In(1)); | ||
| builder.From(s3.Outlet).To(merge.In(2)); | ||
|
|
||
| builder.From(merge.Out).To(sink); |
There was a problem hiding this comment.
Scala code uses just Sink.FromSubscriber(probe), but it does not compile on C#.
alexvaluyskiy
force-pushed
the
mergeprioritized
branch
from
f6e5ee2 to
6e5dac3
Compare
src/core/Akka.Streams/Dsl/Graph.cs
Outdated
| /// Create a new <see cref="T:Akka.Streams.Dsl.MergePrioritized`1" /> with specified number of input ports. | ||
| /// </summary> | ||
| /// <param name="priorities">Priorities of the input ports</param> | ||
| public MergePrioritized(IList<int> priorities) |
There was a problem hiding this comment.
Should I change it to IEnumerable<int>? Originally it was Seq[Int]
There was a problem hiding this comment.
I would say yes, no need for a list since no add/remove operations are needed.
| private RunnableGraph<T> ThreeSourceMerge<T>(Source<T, NotUsed> source1, Source<T, NotUsed> source2, | ||
| Source<T, NotUsed> source3, List<int> priorities, TestSubscriber.ManualProbe<T> probe) | ||
| { | ||
| return RunnableGraph.FromGraph(GraphDsl.Create(source1, source2, source3, (p1, p2, p3) => default(T), (builder, s1, s2, s3) => |
There was a problem hiding this comment.
I think I should change (p1, p2, p3) => default(T) to Tuple.Create
| /// <param name="ops">TBD</param> | ||
| /// <param name="sink">TBD</param> | ||
| /// <returns>TBD</returns> | ||
| public static GraphDsl.Builder<TMat> To<TIn, TOut, TMat>(this GraphDsl.ForwardOps<TOut, TMat> ops, IGraph<SinkShape<TIn>, TMat> sink) | ||
| public static GraphDsl.Builder<TMat> To<TIn, TOut, TMat, TMat2>(this GraphDsl.ForwardOps<TOut, TMat> ops, IGraph<SinkShape<TIn>, TMat2> sink) |
There was a problem hiding this comment.
Scala DSL does not care about the materializer value of the sink. We should do the same
alexvaluyskiy
force-pushed
the
mergeprioritized
branch
from
6a90b30 to
db59cd9
Compare
alexvaluyskiy
changed the title
src/core/Akka.Streams/Dsl/Graph.cs
Outdated
| @@ -424,6 +424,188 @@ public MergePreferred(int secondaryPorts, bool eagerClose = false) | |||
| } | |||
|
|
|||
| /// <summary> | |||
| /// Merge several streams, taking elements as they arrive from input streams | |||
| /// (picking from prioritized once when several have elements ready). | |||
| /// A <see cref="T:Akka.Streams.Dsl.MergePrioritized`1" /> has one <see cref="P:Akka.Streams.Dsl.MergePrioritized`1.Out" /> port, one or more input port with their priorities. | |||
There was a problem hiding this comment.
Is this a new format ?
There was a problem hiding this comment.
What do you mean?
There was a problem hiding this comment.
A <see cref="T:Akka.Streams.Dsl.MergePrioritized`1" /> the only format I know is <see cref="GraphStage{TShape}"/>
src/core/Akka.Streams/Dsl/Graph.cs
Outdated
| /// </summary> | ||
| public sealed class MergePrioritized<T> : GraphStage<UniformFanInShape<T, T>> | ||
| { | ||
| public IList<int> Priorities { get; } |
There was a problem hiding this comment.
I wouldn't expose IList, one could add / remove priorities afterwards
src/core/Akka.Streams/Dsl/Graph.cs
Outdated
| public MergePrioritizedLogic(MergePrioritized<T> stage) : base(stage.Shape) | ||
| { | ||
| _stage = stage; | ||
| allBuffers = new List<FixedSizeBuffer<Inlet<T>>>(); |
There was a problem hiding this comment.
Wouldn't a Array be better since we know the size?
| /// <param name="eagerComplete">If true, the merge will complete as soon as one of its inputs completes</param> | ||
| /// <exception cref="ArgumentException"> | ||
| /// This exception is thrown when the specified <paramref name="priorities"/> is less or equal zero. | ||
| /// </exception> |
There was a problem hiding this comment.
Exception should be also visible on the other constructor or maybe set eagerComplete to false as default and use single constructor
src/core/Akka.Streams/Dsl/Graph.cs
Outdated
|
|
||
| public override UniformFanInShape<T, T> Shape { get; } | ||
|
|
||
| public List<Inlet<T>> In { get; } |
There was a problem hiding this comment.
Should be read only list
|
API approval is missing |
#3110