Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

StreamRefs #3321

Merged
merged 10 commits into from
Feb 24, 2018
Merged

StreamRefs #3321

merged 10 commits into from
Feb 24, 2018

Conversation

Horusiath
Copy link
Contributor

@Horusiath Horusiath commented Feb 10, 2018 •
edited
Loading

This is work in progress over porting StreamRefs feature to .NET.

TODOs:

  • Ported settings, and all stream-ref related types
  • StreamRefSpec (except serialization)
  • Serialization (in progress) - this is the most problematic part, thanks to awesome support of generic data types by Protobuf. Also I've noticed, that JVM implementation gave up on serializing exceptions. I think we could go a little more ambitious here.
  • Documentation

This PR may require akka.streams 2.5 to be merged first (cc: @marcpiechura).

@marcpiechura
Copy link
Contributor

@Horusiath anything in particular ?
Progress on 2.5 is tracked here #2601 (comment) but I‘ve moved most of the tasks to #2552 since they depend on the new materializer

@Horusiath
Copy link
Contributor Author

@marcpiechura I think that there were quite big changes in a way how stage actors are created. From what I've seen, it's no longer getStageActorRef()

@marcpiechura
Copy link
Contributor

@Horusiath I think it’s this PR akka/akka#19487 but it’s quite old, not sure why we haven’t implemented it already

@Horusiath Horusiath mentioned this pull request Feb 14, 2018
Merged
@Horusiath Horusiath changed the base branch from dev to v1.4 February 21, 2018 21:25
@Horusiath
Copy link
Contributor Author

If this CI will pass, this is ready for review and merge. I haven't implemented exception stack trace propagation over the wire, but it can be added even after the release.

@marcpiechura
Copy link
Contributor

@Horusiath looks like you forgot to set the config parameter in the reference configuration for the tests ?!

System.ArgumentNullException : `akka.stream.materializer.stream-ref` was not present
Parameter name: config
   at Akka.Streams.Implementation.StreamRefSettings.Create(Config config)
   at Akka.Streams.ActorMaterializerSettings.Create(Config config)
   at Akka.Streams.Tests.Dsl.ActorRefSourceSpec..ctor(

@Horusiath Horusiath changed the title [WIP] StreamRefs StreamRefs Feb 24, 2018
@Horusiath
Copy link
Contributor Author

@marcpiechura for some reason, when no config is provided, a reference.conf from Akka.Streams is not applied.

@marcpiechura
Copy link
Contributor

@Horusiath One spec is still failing

StreamRefsSpec.SinkRef_must_receive_hundreds_of_elements_via_remoting
Failed: Expected a message of type System.String, but received {Failure: System.ArgumentException: Cannot push port twice SourceRef.out

@Horusiath
Copy link
Contributor Author

It's racy behavior, working on it.

marcpiechura
marcpiechura suggested changes Feb 24, 2018
View reviewed changes
docs/articles/streams/streamrefs.md Outdated

The process of preparing and running a `ISourceRef<T>` powered distributed stream is shown by the animation below:

![]()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing link to the animations

docs/articles/streams/streamrefs.md Outdated

The process of preparing and running a `ISinkRef<>` powered distributed stream is shown by the animation below:

![]()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As above

src/core/Akka.Streams/Dsl/StreamRefs.cs Outdated
{
public CumulativeDemand(long seqNr)
{
if (seqNr < 0) throw ReactiveStreamsCompliance.NumberOfElementsInRequestMustBePositiveException;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should be <=

src/core/Akka.Streams/Dsl/StreamRefs.cs
private readonly EnumerableActorName sourceRefStageNames = new EnumerableActorNameImpl("SourceRef", new AtomicCounterLong(0L));
private readonly EnumerableActorName sinkRefStageNames = new EnumerableActorNameImpl("SinkRef", new AtomicCounterLong(0L));

public StreamRefsMaster(ExtendedActorSystem system)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can the parameter be removed since it’s not stored ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just want to keep the convention for extensions here.

src/core/Akka.Streams/Dsl/StreamRefs.cs
internal sealed class SourceRefImpl<T> : SourceRefImpl, ISourceRef<T>
{
public SourceRefImpl(IActorRef initialPartnerRef) : base(initialPartnerRef) { }
public override Type EventType => typeof(T);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wouldn’t be a single assignment better ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's used only for deserialization (usually once per instance per node). I don't want to allocate underlying field for it, as Hyperion/Json.NET are able to provide generic type param directly from SourceRefImpl<T>'s T parameter, so the payload will be smaller.

src/core/Akka.Streams/Dsl/StreamRefs.cs
internal sealed class SinkRefImpl<T> : SinkRefImpl, ISinkRef<T>
{
public SinkRefImpl(IActorRef initialPartnerRef) : base(initialPartnerRef) { }
public override Type EventType => typeof(T);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As above

src/core/Akka.Streams/Dsl/StreamRefs.cs Outdated
this.SetHandler(_stage.Inlet,
onPush: OnPush,
onUpstreamFinish: OnUpstreamFinish,
onUpstreamFailure: OnUpstreamFailure);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Implement IInHandler and use SetHandler(_stage.Inlet, this)

src/core/Akka.Streams/Dsl/StreamRefs.cs Outdated
_promise = promise;
_inheritedAttributes = inheritedAttributes;

SetHandler(_stage.Outlet, onPull: OnPull);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Implement IOutHandler

marcpiechura
marcpiechura reviewed Feb 24, 2018
View reviewed changes
src/core/Akka.Streams/Dsl/StreamRefs.cs Outdated

public void OnDownstreamFinish()
{
/* IOutHandler impl */
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Needs to call CompleteStage

@marcpiechura marcpiechura merged commit 184e7c7 into akkadotnet:v1.4 Feb 24, 2018
@marcpiechura marcpiechura mentioned this pull request Mar 23, 2018
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@marcpiechura marcpiechura marcpiechura approved these changes

Assignees
No one assigned
Labels
needs review
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.

None yet
3 participants
@Horusiath @marcpiechura @Aaronontheweb