Adding placeholders for time-based Skip and Take.

AsyncRx.NET/System.Reactive.Async/System/Reactive/Linq/Operators/Skip.cs

@@ -2,6 +2,8 @@
 // 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.Reactive.Concurrency;
+
 namespace System.Reactive.Linq
 {
     partial class AsyncObservable
@@ -20,6 +22,38 @@ namespace System.Reactive.Linq
 
             return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Skip(observer, count)));
         }
+
+        public static IAsyncObservable<TSource> Skip<TSource>(this IAsyncObservable<TSource> source, TimeSpan duration)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+
+            if (duration == TimeSpan.Zero)
+            {
+                return source;
+            }
+
+            return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Skip(observer, duration)));
+        }
+
+        public static IAsyncObservable<TSource> Skip<TSource>(this IAsyncObservable<TSource> source, TimeSpan duration, IAsyncScheduler scheduler)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+            if (scheduler == null)
+                throw new ArgumentNullException(nameof(scheduler));
+
+            if (duration == TimeSpan.Zero)
+            {
+                return source;
+            }
+
+            return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Skip(observer, duration, scheduler)));
+        }
     }
 
     partial class AsyncObserver
@@ -47,5 +81,27 @@ namespace System.Reactive.Linq
                 observer.OnCompletedAsync
             );
         }
+
+        public static IAsyncObserver<TSource> Skip<TSource>(IAsyncObserver<TSource> observer, TimeSpan duration)
+        {
+            if (observer == null)
+                throw new ArgumentNullException(nameof(observer));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+
+            throw new NotImplementedException();
+        }
+
+        public static IAsyncObserver<TSource> Skip<TSource>(IAsyncObserver<TSource> observer, TimeSpan duration, IAsyncScheduler scheduler)
+        {
+            if (observer == null)
+                throw new ArgumentNullException(nameof(observer));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+            if (scheduler == null)
+                throw new ArgumentNullException(nameof(scheduler));
+
+            throw new NotImplementedException();
+        }
     }
 }

AsyncRx.NET/System.Reactive.Async/System/Reactive/Linq/Operators/Take.cs

@@ -2,6 +2,8 @@
 // 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.Reactive.Concurrency;
+
 namespace System.Reactive.Linq
 {
     partial class AsyncObservable
@@ -20,6 +22,38 @@ namespace System.Reactive.Linq
 
             return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Take(observer, count)));
         }
+
+        public static IAsyncObservable<TSource> Take<TSource>(this IAsyncObservable<TSource> source, TimeSpan duration)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+
+            if (duration == TimeSpan.Zero)
+            {
+                return Empty<TSource>();
+            }
+
+            return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Take(observer, duration)));
+        }
+
+        public static IAsyncObservable<TSource> Take<TSource>(this IAsyncObservable<TSource> source, TimeSpan duration, IAsyncScheduler scheduler)
+        {
+            if (source == null)
+                throw new ArgumentNullException(nameof(source));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+            if (scheduler == null)
+                throw new ArgumentNullException(nameof(scheduler));
+
+            if (duration == TimeSpan.Zero)
+            {
+                return Empty<TSource>();
+            }
+
+            return Create<TSource>(observer => source.SubscribeAsync(AsyncObserver.Take(observer, duration, scheduler)));
+        }
     }
 
     partial class AsyncObserver
@@ -47,5 +81,27 @@ namespace System.Reactive.Linq
                 observer.OnCompletedAsync
             );
         }
+
+        public static IAsyncObserver<TSource> Take<TSource>(IAsyncObserver<TSource> observer, TimeSpan duration)
+        {
+            if (observer == null)
+                throw new ArgumentNullException(nameof(observer));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+
+            throw new NotImplementedException();
+        }
+
+        public static IAsyncObserver<TSource> Take<TSource>(IAsyncObserver<TSource> observer, TimeSpan duration, IAsyncScheduler scheduler)
+        {
+            if (observer == null)
+                throw new ArgumentNullException(nameof(observer));
+            if (duration < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(duration));
+            if (scheduler == null)
+                throw new ArgumentNullException(nameof(scheduler));
+
+            throw new NotImplementedException();
+        }
     }
 }