AsyncEnumerable by eaba · Pull Request #6044 · akkadotnet/akka.net

eaba · 2022-07-13T23:59:31Z

Changes

IAsyncEnumerable as a streams source

to11mtm · 2022-07-14T16:54:53Z

+            //TODO: when to dispose async enumerable? Should this be a part of ownership of current stage, or should it
+            // be a responsibility of the caller?
+            _asyncEnumerable = asyncEnumerable;


@Aaronontheweb :

Not entirely related but Should we be taking a Func<IAsyncEnumerable<T>> here? (In my opinion yes, in ideal-state, a Source itself should be idempotent and able to be materialized multiple times.)

As to this question, for whatever my opinion is worth I'd think that since we are consuming the stream till the end, we should be disposing when we are done whether it is provided by a func producer or live instance.

Not entirely related but Should we be taking a Func<IAsyncEnumerable> here? (In my opinion yes, in ideal-state, a Source itself should be idempotent and able to be materialized multiple times.)

Now that is a good question. We should absolutely program this component to accept this Func so we can restart the stream in the event of errors. That'll also make disposal of the IAsyncEnumerable less ambigious too: we created it, we own it, so we clean it up.

@eaba can you implement this change?

@to11mtm Func<IAsyncEnumerable>

ty_IAsyncEnumerable_Is_Passed_In
Source: AsyncEnumerableSpec.cs line 163
Duration: 1.7 sec

Message:
Failed: Expected a message of type Akka.Streams.TestKit.TestSubscriber+OnComplete, but received {TestSubscriber.OnNext(21)} (type Akka.Streams.TestKit.TestSubscriber+OnNext`1[System.Int32]) instead from [akka://AsyncEnumerableSpec-1/user/StreamSupervisor-0/Flow-0-0-enumerableSource#1834516904]
Expected: True
Actual: False

Stack Trace:
XunitAssertions.Fail(String format, Object[] args) line 26
d__1361.MoveNext() line 455 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ValueTask1.get_Result()
ConfiguredValueTaskAwaiter.GetResult()
d__1351.MoveNext() line 421 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ValueTask1.get_Result()
ConfiguredValueTaskAwaiter.GetResult()
d__1331.MoveNext() line 389 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ValueTask1.get_Result()
ConfiguredValueTaskAwaiter.GetResult()
d__1311.MoveNext() line 367 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ValueTask1.get_Result()
ConfiguredValueTaskAwaiter.GetResult()
d__1151.MoveNext() line 52 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ValueTask1.get_Result()
ValueTaskAwaiter1.GetResult() <ExpectCompleteTask>d__85.MoveNext() line 117 --- End of stack trace from previous location where exception was thrown --- TaskAwaiter.ThrowForNonSuccess(Task task) TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task) ConfiguredTaskAwaiter.GetResult() ManualProbe1.ExpectComplete(CancellationToken cancellationToken) line 186
AsyncEnumerableSpec.AsyncEnumerableSource_Must_Complete_Immediately_With_No_elements_When_An_Empty_IAsyncEnumerable_Is_Passed_In() line 182

Standard Output:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
[DEBUG][7/16/2022 2:09:57 PM][Thread 0027][ActorSystem(AsyncEnumerableSpec-1)] System shutdown initiated
[DEBUG][7/16/2022 2:09:57 PM][Thread 0020][EventStream] Shutting down: StandardOutLogger started
[DEBUG][7/16/2022 2:09:57 PM][Thread 0020][EventStream] All default loggers stopped

Aaronontheweb · 2022-07-15T14:16:53Z

the code in this PR is copied from @Horusiath Akka.Persistence.Pulsar - and while the copyright headers and licenses are identical to Akka.NET's we'll need to obtain his permission before proceeding.

Horusiath · 2022-07-15T14:22:35Z

@Aaronontheweb @eaba I don't mind coping the code. Feel free to use it.

Aaronontheweb · 2022-07-15T14:56:38Z

thanks @Horusiath - we'll move forward with these changes then.

Aaronontheweb · 2022-07-15T15:02:02Z

Once this is merged we'll need to backport it to the v1.4 branch in order to close #5836

Aaronontheweb · 2022-07-18T11:31:10Z

Compilation error on Linux...

/home/vsts/work/1/s/src/core/Akka.Streams.Tests/Dsl/AsyncEnumerableSpec.cs(185,22): error CS0433: The type 'IAsyncEnumerable<T>' exists in both 'Microsoft.Bcl.AsyncInterfaces, Version=6.0.0.0, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51' and 'mscorlib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089' [/home/vsts/work/1/s/src/core/Akka.Streams.Tests/Akka.Streams.Tests.csproj]
    557 Warning(s)
    1 Error(s)

Aaronontheweb · 2022-07-18T11:31:35Z

Weird that we'd have that problem given that we already have IAsyncEnumerable support elsewhere in Akka.Streams

Aaronontheweb · 2022-07-18T16:15:48Z

Current test failures are unrelated to your PR I think - looks like mostly Akka.Persistence issues.

Aaronontheweb

LGTM

Aaronontheweb · 2022-07-18T16:34:02Z

    [Akka.Annotations.InternalApiAttribute()]
+    public sealed class AsyncEnumerable<T> : Akka.Streams.Stage.GraphStage<Akka.Streams.SourceShape<T>>
+    {
+        public AsyncEnumerable(System.Func<System.Collections.Generic.IAsyncEnumerable<T>> factory) { }


Aaronontheweb · 2022-07-18T16:35:20Z

+                }
+                else
+                {
+                    vtask.AsTask().ContinueWith(_handleContinuation);


This would probably be better with a local function that does await but we can refactor this later.

* Cherry-picked from af513b0 * Fix missing DefaultConfig * Fix missing StreamTestDefaultMailbox.DefaultConfig Co-authored-by: Ebere Abanonu <eaba@users.noreply.github.com>

eaba added 8 commits July 14, 2022 00:53

AsyncEnumerable

3eb9e9f

AsyncEnumerableSpec

1196a55

API Tests

d75207f

API Test NETCore

626d0cd

AkkaSpecConfig

622682b

#if NETCOREAPP

24fee71

Horusiath AsyncEnumerableSource @Horusiath

8936f11

API Tests

abec092

to11mtm reviewed Jul 14, 2022

View reviewed changes

Aaronontheweb added the akka-streams label Jul 15, 2022

Aaronontheweb added this to the 1.5.0 milestone Jul 15, 2022

eaba added 2 commits July 16, 2022 14:15

Func<IAsyncEnumerable<T>>

f96b5eb

Core API Spec

7ef8b91

Aaronontheweb reviewed Jul 18, 2022

View reviewed changes

Comment thread src/core/Akka.Streams.Tests/Dsl/AsyncEnumerableSpec.cs

NETCOREAPP

2d35faf

Aaronontheweb approved these changes Jul 18, 2022

View reviewed changes

Update Ops.cs

2eea057

Aaronontheweb enabled auto-merge (squash) July 18, 2022 16:43

Aaronontheweb disabled auto-merge July 18, 2022 16:43

Aaronontheweb enabled auto-merge (squash) July 18, 2022 16:43

Aaronontheweb merged commit af513b0 into akkadotnet:dev Jul 18, 2022

eaba deleted the async_enumerable_as_run branch July 18, 2022 17:17

Aaronontheweb mentioned this pull request Jul 18, 2022

cleaned up IAsyncEnumerable Source to use local functions #6045

Merged

3 tasks done

Arkatufus mentioned this pull request Jul 21, 2022

[BACKPORT #6044] Add IAsyncEnumerable as Akka.Streams Source #6047

Merged

Conversation

eaba commented Jul 13, 2022

Changes

Uh oh!

to11mtm Jul 14, 2022

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb Jul 15, 2022

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb Jul 15, 2022

Choose a reason for hiding this comment

Uh oh!

eaba Jul 16, 2022

Choose a reason for hiding this comment

Uh oh!

eaba Jul 16, 2022

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb commented Jul 15, 2022

Uh oh!

Horusiath commented Jul 15, 2022

Uh oh!

Aaronontheweb commented Jul 15, 2022

Uh oh!

Aaronontheweb commented Jul 15, 2022

Uh oh!

Aaronontheweb commented Jul 18, 2022

Uh oh!

Aaronontheweb commented Jul 18, 2022

Uh oh!

Uh oh!

Aaronontheweb commented Jul 18, 2022

Uh oh!

Aaronontheweb left a comment

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb Jul 18, 2022

Choose a reason for hiding this comment

Uh oh!

Aaronontheweb Jul 18, 2022

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants