Adding AsyncObserverBase<T>.

AsyncRx.NET/System.Reactive.Async/System/Reactive/Linq/AsyncObserver.cs

@@ -20,7 +20,7 @@ namespace System.Reactive.Linq
             return new AnonymousAsyncObserver<T>(onNextAsync, onErrorAsync, onCompletedAsync);
         }
 
-        private sealed class AnonymousAsyncObserver<T> : IAsyncObserver<T>
+        private sealed class AnonymousAsyncObserver<T> : AsyncObserverBase<T>
         {
             private readonly Func<T, Task> _onNextAsync;
             private readonly Func<Exception, Task> _onErrorAsync;
@@ -33,20 +33,11 @@ namespace System.Reactive.Linq
                 _onCompletedAsync = onCompletedAsync;
             }
 
-            public Task OnCompletedAsync()
-            {
-                throw new NotImplementedException();
-            }
+            protected override Task OnCompletedAsyncCore() => _onCompletedAsync();
 
-            public Task OnErrorAsync(Exception error)
-            {
-                throw new NotImplementedException();
-            }
+            protected override Task OnErrorAsyncCore(Exception error) => _onErrorAsync(error);
 
-            public Task OnNextAsync(T value)
-            {
-                throw new NotImplementedException();
-            }
+            protected override Task OnNextAsyncCore(T value) => _onNextAsync(value);
         }
     }
 }

AsyncRx.NET/System.Reactive.Async/System/Reactive/Linq/AsyncObserverBase.cs

@@ -0,0 +1,35 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information. 
+
+using System.Threading.Tasks;
+
+namespace System.Reactive.Linq
+{
+    public abstract class AsyncObserverBase<T> : IAsyncObserver<T>
+    {
+        public Task OnCompletedAsync()
+        {
+            throw new NotImplementedException();
+        }
+
+        protected abstract Task OnCompletedAsyncCore();
+
+        public Task OnErrorAsync(Exception error)
+        {
+            if (error == null)
+                throw new ArgumentNullException(nameof(error));
+
+            throw new NotImplementedException();
+        }
+
+        protected abstract Task OnErrorAsyncCore(Exception error);
+
+        public Task OnNextAsync(T value)
+        {
+            throw new NotImplementedException();
+        }
+
+        protected abstract Task OnNextAsyncCore(T value);
+    }
+}