-
Notifications
You must be signed in to change notification settings - Fork 1k
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.
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
Move Channel Stages from Alpakka to main project. #6268
Move Channel Stages from Alpakka to main project. #6268
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Left my own comments here.
Please let me know if files should be moved/deleted/etc.
CompleteStage(); | ||
} | ||
else | ||
continuation.AsTask().ContinueWith(_onReadReady); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is a slight change from the Alpakka code:
- The Alpakka code looked a little racy with how it was checking flags.
- The Failure/Completion stages are less perf sensitive than the main read loop, so the code on our happy path is now smaller
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
_onValueReadFailure = | ||
GetAsyncCallback<Exception>(OnValueReadFailure); | ||
_onReaderComplete = GetAsyncCallback<Exception>(OnReaderComplete); | ||
_onReadReady = ContinueAsyncRead; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This change is here to avoid delegates potentially being made/initted repeatedly in OnPull.
/// <param name="singleReader"></param> | ||
/// <param name="fullMode"></param> | ||
/// <returns></returns> | ||
public static Sink<T, ChannelReader<T>> AsReader<T>(int bufferSize, bool singleReader = false, BoundedChannelFullMode fullMode = BoundedChannelFullMode.Wait) => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IDK if we want to keep these, I moved them over for the sake of migration.
/// <typeparam name="T"></typeparam> | ||
/// <param name="reader"></param> | ||
/// <returns></returns> | ||
public static Source<T, NotUsed> FromReader<T>(ChannelReader<T> reader) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IDK if we want to keep these, I moved them over for the sake of migration.
bool singleWriter = false, | ||
BoundedChannelFullMode fullMode = BoundedChannelFullMode.Wait) | ||
{ | ||
return Source.FromGraph( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I added this for the sake of symmetry with the Sinks, as well as it's amazing utility in place of Source.Queue<T>
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Left some thoughts
public ChannelReaderSink(int bufferSize, bool singleReader = false, BoundedChannelFullMode fullMode = BoundedChannelFullMode.Wait) | ||
{ | ||
_bufferSize = bufferSize; | ||
_singleReader = singleReader; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't singleReader
be true
by default?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This was 'as-found' but singleReader = false
here is the safer default so I didn't think it was worth changing.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
FWIW, BoundedChannel
AFAIK doesn't even use ChannelOptions.SingleReader
. it's only used in UnboundedChannel
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And we don't use UnboundedChannel
, so I guess this setting doesn't really matter then?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And we don't use
UnboundedChannel
, so I guess this setting doesn't really matter then?
Not yet anyway. The challenge of course is that it may become relevant in the future.
Right now, BoundedChannel<T>
uses a Dequeue<T>
internally, so it uses a single lock for both readers/writers. In the future however, the internal structure may change (e.x. something more like ConcurrentQueue<T>
with segments) and then it may become important.
Really it's a question of whether it is an oversight in ChannelOptions
that it is exposed for both types.
CompleteStage(); | ||
} | ||
else | ||
continuation.AsTask().ContinueWith(_onReadReady); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
We can add website documentation later, but I think this looks good - can you run API approvals @to11mtm ? |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM but needs API approvals
I think I have the API approvals working here now |
Looks like one of the new versions of Verify we started using might be causing problems with API approval tests. |
* Move Channel Stages from Alpakka to main project. * added API approvals Co-authored-by: Aaron Stannard <[email protected]>
) * Move Channel Stages from Alpakka to main project. (#6268) * Move Channel Stages from Alpakka to main project. * added API approvals Co-authored-by: Aaron Stannard <[email protected]> * added API approvals Co-authored-by: Drew <[email protected]>
Changes
Source
andSink
to have Channel methodsChannelWriter<T>
resultSource.Queue<T>
.The main motivations for this are as follows:
System.Threading.Channels
as a dep in core AkkaRunAsAsyncEnumerable
, even if it is a minorly-breaking API change it may be better long term.Checklist
For significant changes, please ensure that the following have been completed (delete if not relevant):