Resuming client stream subscription about orleankka HOT 12 CLOSED

 static class StreamSubscriptionExtensions
    {
        private static readonly FieldInfo handle = typeof(StreamSubscription).GetTypeInfo().GetDeclaredField("handle");
        private static readonly MethodInfo GetAllSubscriptionHandles = typeof(StreamRef).GetTypeInfo().GetDeclaredMethod("GetAllSubscriptionHandles");

        /// <summary>
        ///   Returns a list of all current stream subscriptions.
        /// </summary>
        /// <returns> A promise for a list of StreamSubscription </returns>
        public static async Task<List<StreamSubscription>> Subscriptions(this StreamRef streamRef)
        {
            var handles = await (Task<IList<StreamSubscriptionHandle<object>>>)GetAllSubscriptionHandles.Invoke(streamRef, null);

            return handles.Select(CreateStreamSubscription).ToList();
        }

        private static StreamSubscription CreateStreamSubscription(StreamSubscriptionHandle<object> x)
        {
            return Activator.CreateInstance(typeof(StreamSubscription), BindingFlags.Instance | BindingFlags.NonPublic, null, new object[] { x }, null, null) as StreamSubscription;
        }

        public static Task Resume(this StreamSubscription streamSubscription, Func<object, Task> callback)
        {
            Requires.NotNull(callback, nameof(callback));
            var observer = new Observer((item, token) => callback(item));
            var subscriptionHandle = (StreamSubscriptionHandle<object>)handle.GetValue(streamSubscription);

            return subscriptionHandle.ResumeAsync(observer);
        }

        public static Task Resume<T>(this StreamSubscription streamSubscription, Func<T, Task> callback)
        {
            Requires.NotNull(callback, nameof(callback));
            return streamSubscription.Resume(item => callback((T)item));
        }

        class Observer : IAsyncObserver<object>
        {
            readonly Func<object, StreamSequenceToken, Task> callback;

            public Observer(Func<object, StreamSequenceToken, Task> callback)
            {
                this.callback = callback;
            }

            public Task OnNextAsync(object item, StreamSequenceToken token = null)
                => callback(item, token);

            public Task OnCompletedAsync() => Task.CompletedTask;
            public Task OnErrorAsync(Exception ex) => Task.CompletedTask;
        }
    }