delay and worker Observers

Sources/Disposable.swift

@@ -53,22 +53,6 @@ public final class CompositeDisposable: Disposable {
     }
 }
 
-public final class SerialDisposable: Disposable {
-    private var disposable: Disposable?
-
-    public init() {
-    }
-
-    public func swap(with: Disposable) {
-        disposable = with
-    }
-
-    public func dispose() {
-        disposable?.dispose()
-        disposable = nil
-    }
-}
-
 public final class SwapableDisposable: Disposable {
     private var parentDisposable: Disposable?
     private var childDisposable: Disposable?

Sources/Observable+Task.swift

@@ -263,7 +263,7 @@ extension Observable where ValueType: ResultConvertible {
         var numberOfRetries = 0
 
         return Task<ResultValueType> { completion  in
-            let serial = SerialDisposable()
+            let serial = SwapableDisposable()
 
             func retryImpl() -> Disposable {
                 return self.start { result in
@@ -273,8 +273,8 @@ extension Observable where ValueType: ResultConvertible {
                     case .failure(let error):
                         numberOfRetries += 1
                         if until(error) && (numberOfRetries <= numberOfTimes) {
-                            serial.swap(with: queue.after(timeInterval: currentTimeout) {
-                                serial.swap(with: retryImpl())
+                            serial.swap(child: queue.after(timeInterval: currentTimeout) {
+                                serial.swap(parent: retryImpl())
                             })
                             currentTimeout = nextTimeout(currentTimeout)
                         } else {
@@ -284,7 +284,7 @@ extension Observable where ValueType: ResultConvertible {
                 }
             }
 
-            serial.swap(with: retryImpl())
+            serial.swap(parent: retryImpl())
             return serial
         }
     }

Sources/Observer.swift

@@ -25,6 +25,27 @@ public struct Observer<T>: Observable {
 }
 
 extension Observer {
+
+    /**
+     Create task with worker which will be run in workerQueue and send result to completionQueue. Worker can produce value or error.
+     */
+    public init(workerQueue: DispatchQueue, completionQueue: DispatchQueue = .main, worker: @escaping () -> ValueType) {
+        self.init { completion in
+            return workerQueue.run(worker: worker, completionQueue: completionQueue) { (value: ValueType) in
+                completion(value)
+            }
+        }
+    }
+
+    public static func delay(timeInterval: TimeInterval, queue: DispatchQueue = .main) ->  Observer<Void> {
+        return Observer<Void> { observer in
+            return queue.after(timeInterval: timeInterval) {
+                observer(())
+            }
+        }
+    }
+
+
     public static func from(value: T) -> Observer<T> {
         return Observer<T> { observer in
             observer(value)