Handle And filter expired tasks

masmovil · Dec 1, 2023 · 9300e78 · 9300e78
1 parent 488b802
commit 9300e78
Show file tree

Hide file tree

Showing 8 changed files with 147 additions and 3 deletions.
diff --git a/Mini.xcodeproj/project.pbxproj b/Mini.xcodeproj/project.pbxproj
@@ -141,6 +141,10 @@
 		F297D27F286A0C6900323F24 /* Dispatcher+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297D27D286A0C6900323F24 /* Dispatcher+Combine.swift */; };
 		F297D280286A0C6900323F24 /* Dispatcher+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297D27D286A0C6900323F24 /* Dispatcher+Combine.swift */; };
 		F297D281286A0C6900323F24 /* Dispatcher+Combine.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297D27D286A0C6900323F24 /* Dispatcher+Combine.swift */; };
+		F297F4522B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */; };
+		F297F4532B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */; };
+		F297F4542B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */; };
+		F297F4552B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */ = {isa = PBXBuildFile; fileRef = F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */; };
 		F2AD8249286B6AD9005C024F /* TaskExpiration.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2AD8248286B6AD9005C024F /* TaskExpiration.swift */; };
 		F2AD824A286B6AD9005C024F /* TaskExpiration.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2AD8248286B6AD9005C024F /* TaskExpiration.swift */; };
 		F2AD824B286B6AD9005C024F /* TaskExpiration.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2AD8248286B6AD9005C024F /* TaskExpiration.swift */; };
@@ -250,6 +254,7 @@
 		F288DCD829BB942100FBFED1 /* Publishers.CombineMiniTasksArray.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Publishers.CombineMiniTasksArray.swift; sourceTree = "<group>"; };
 		F297D268286A02E200323F24 /* KeyedAction.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = KeyedAction.swift; sourceTree = "<group>"; };
 		F297D27D286A0C6900323F24 /* Dispatcher+Combine.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "Dispatcher+Combine.swift"; sourceTree = "<group>"; };
+		F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Publishers.RemoveExpired.swift; sourceTree = "<group>"; };
 		F2AD8248286B6AD9005C024F /* TaskExpiration.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TaskExpiration.swift; sourceTree = "<group>"; };
 		F2AD824D286B7065005C024F /* PublishersTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = PublishersTests.swift; sourceTree = "<group>"; };
 		F2C09DAB286B1490009C9C8E /* TestError.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TestError.swift; sourceTree = "<group>"; };
@@ -441,6 +446,7 @@
 				F288DCD329BB922F00FBFED1 /* Publishers.CombineMiniTasksTuple3.swift */,
 				F288DCCE29BB922700FBFED1 /* Publishers.CombineMiniTasksTuple4.swift */,
 				3A20F7E129CCAC5500DDCF8D /* Publishers.EraseToEmptyTask.swift */,
+				F297F4512B1A313200B7B1FA /* Publishers.RemoveExpired.swift */,
 			);
 			path = Publishers;
 			sourceTree = "<group>";
@@ -802,6 +808,7 @@
 				F222D4F525249B9B00672E7B /* DispatchQueueExtensions.swift in Sources */,
 				F2D0DA1D29E8473700A114EC /* EmptyTask.swift in Sources */,
 				F222D4BF25249B7E00672E7B /* ActionReducer.swift in Sources */,
+				F297F4522B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */,
 				F26C3AF822537A4600189D28 /* Mini.swift in Sources */,
 				F2AD8249286B6AD9005C024F /* TaskExpiration.swift in Sources */,
 				F2D0DA1829E8472900A114EC /* Taskable.swift in Sources */,
@@ -861,6 +868,7 @@
 				F222D4F625249B9B00672E7B /* DispatchQueueExtensions.swift in Sources */,
 				F2D0DA1E29E8473700A114EC /* EmptyTask.swift in Sources */,
 				F222D4C025249B7E00672E7B /* ActionReducer.swift in Sources */,
+				F297F4532B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */,
 				F26C3AF922537A4600189D28 /* Mini.swift in Sources */,
 				F2AD824A286B6AD9005C024F /* TaskExpiration.swift in Sources */,
 				F2D0DA1929E8472900A114EC /* Taskable.swift in Sources */,
@@ -920,6 +928,7 @@
 				F222D4F725249B9B00672E7B /* DispatchQueueExtensions.swift in Sources */,
 				F2D0DA1F29E8473700A114EC /* EmptyTask.swift in Sources */,
 				F222D4C125249B7E00672E7B /* ActionReducer.swift in Sources */,
+				F297F4542B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */,
 				F26C3AFA22537A4600189D28 /* Mini.swift in Sources */,
 				F2AD824B286B6AD9005C024F /* TaskExpiration.swift in Sources */,
 				F2D0DA1A29E8472900A114EC /* Taskable.swift in Sources */,
@@ -979,6 +988,7 @@
 				F222D4F825249B9B00672E7B /* DispatchQueueExtensions.swift in Sources */,
 				F2D0DA2029E8473700A114EC /* EmptyTask.swift in Sources */,
 				F222D4C225249B7E00672E7B /* ActionReducer.swift in Sources */,
+				F297F4552B1A313200B7B1FA /* Publishers.RemoveExpired.swift in Sources */,
 				F26C3AFB22537A4600189D28 /* Mini.swift in Sources */,
 				F2AD824C286B6AD9005C024F /* TaskExpiration.swift in Sources */,
 				F2D0DA1B29E8472900A114EC /* Taskable.swift in Sources */,

diff --git a/Sources/Publishers/Publishers.EraseToEmptyTask.swift b/Sources/Publishers/Publishers.EraseToEmptyTask.swift
@@ -1,7 +1,7 @@
 import Combine
 
 public extension Publisher {
-    func eraseToEmptyTask() -> Publishers.EraseToEmptyTask<Self, Output.Failure>
+    func eraseToEmptyTask() -> Publishers.EraseToEmptyTask<Self>
     where Output: Taskable {
         Publishers.EraseToEmptyTask(upstream: self)
     }
@@ -10,8 +10,8 @@ public extension Publisher {
 public extension Publishers {
     /// Create a `Publisher` that connect an Upstream (Another publisher) that type erases `Task`s to `EmptyTask`
     /// The Output of this `Publisher` always is a combined `EmptyTask`
-    struct EraseToEmptyTask<Upstream: Publisher, TaskFailure: Error>: Publisher where Upstream.Output: Taskable, Upstream.Output.Failure == TaskFailure {
-        public typealias Output = EmptyTask<TaskFailure>
+    struct EraseToEmptyTask<Upstream: Publisher>: Publisher where Upstream.Output: Taskable {
+        public typealias Output = EmptyTask<Upstream.Output.Failure>
         public typealias Failure = Upstream.Failure
 
         public let upstream: Upstream

diff --git a/Sources/Publishers/Publishers.RemoveExpired.swift b/Sources/Publishers/Publishers.RemoveExpired.swift
@@ -0,0 +1,54 @@
+import Combine
+
+public extension Publisher {
+    func removeExpired() -> Publishers.RemoveExpired<Self>
+    where Output: Taskable {
+        Publishers.RemoveExpired(upstream: self)
+    }
+}
+
+public extension Publishers {
+    /// Create a `Publisher` that connect an Upstream (Another publisher) that type erases `Task`s to `EmptyTask`
+    /// The Output of this `Publisher` always is a combined `EmptyTask`
+    struct RemoveExpired<Upstream: Publisher>: Publisher where Upstream.Output: Taskable {
+        public typealias Output = Upstream.Output
+        public typealias Failure = Upstream.Failure
+
+        public let upstream: Upstream
+
+        public init(upstream: Upstream) {
+            self.upstream = upstream
+        }
+
+        public func receive<S: Subscriber>(subscriber: S) where Upstream.Failure == S.Failure, Output == S.Input {
+            upstream.subscribe(Inner(downstream: subscriber))
+        }
+    }
+}
+
+extension Publishers.RemoveExpired {
+    private struct Inner<Downstream: Subscriber>: Subscriber
+    where Downstream.Input == Output, Downstream.Failure == Upstream.Failure, Output: Taskable {
+        let combineIdentifier = CombineIdentifier()
+        private let downstream: Downstream
+
+        fileprivate init(downstream: Downstream) {
+            self.downstream = downstream
+        }
+
+        func receive(subscription: Subscription) {
+            downstream.receive(subscription: subscription)
+        }
+
+        func receive(_ input: Upstream.Output) -> Subscribers.Demand {
+            if input.isExpired {
+                return .none
+            }
+            return downstream.receive(input)
+        }
+
+        func receive(completion: Subscribers.Completion<Upstream.Failure>) {
+            downstream.receive(completion: completion)
+        }
+    }
+}
diff --git a/Sources/Task/KeyedTask.swift b/Sources/Task/KeyedTask.swift
@@ -23,6 +23,11 @@ extension KeyedTask where Key: Hashable, Value: Taskable {
         self[key]?.isRunning ?? false
     }
 
+    /// Returns true if the KeyedTask contains a task with given key and its expired. If the key don't exists return false
+    public func isExpired(key: Key) -> Bool {
+        self[key]?.isExpired ?? false
+    }
+
     /// Returns true if the KeyedTask contains a task with given key and its recently succeded. If the key don't exists return false
     public func isRecentlySucceeded(key: Key) -> Bool {
         self[key]?.isRecentlySucceeded ?? false

diff --git a/Sources/Task/Task.swift b/Sources/Task/Task.swift
@@ -50,6 +50,11 @@ public class Task<T: Equatable, E: Error & Equatable>: Taskable, Equatable, Cust
         status == .running
     }
 
+    public var isExpired: Bool {
+        let margin: TimeInterval = 0.1 // 100ms for suscriptions propagations
+        return started.timeIntervalSinceNow + expiration.value + margin < 0
+    }
+
     public var isRecentlySucceeded: Bool {
         switch status {
         case .success where started.timeIntervalSinceNow + expiration.value >= 0:

diff --git a/Sources/Task/Taskable.swift b/Sources/Task/Taskable.swift
@@ -6,6 +6,7 @@ public protocol Taskable {
 
     var isIdle: Bool { get }
     var isRunning: Bool { get }
+    var isExpired: Bool { get }
     var isRecentlySucceeded: Bool { get }
     var isTerminal: Bool { get }
     var isSuccessful: Bool { get }

diff --git a/Tests/PublishersTests.swift b/Tests/PublishersTests.swift
@@ -3,9 +3,11 @@ import Combine
 import XCTest
 
 class PublishersTests: XCTestCase {
+    var taskSuccessExpired: Task<String, TestError> = .success("hola viejo", started: Date() - 1_000, expiration: .immediately)
     var taskSuccess1: Task<String, TestError> = .success("hola")
     var taskSuccess2: Task<String, TestError> = .success("chau")
     var taskFailure1: Task<String, TestError> = .failure(.berenjenaError)
+    var taskFailureExpired: Task<String, TestError> = .failure(.berenjenaError, started: Date() - 1_000)
     var taskFailure2: Task<String, TestError> = .failure(.bigBerenjenaError)
     var taskRunning1: Task<String, TestError> = .running()
     var taskIdle1: Task<String, TestError> = .idle()
@@ -291,4 +293,30 @@ class PublishersTests: XCTestCase {
 
         waitForExpectations(timeout: 2)
     }
+
+    // Remove Expired
+
+    func test_remove_expired() {
+        var cancellables = Set<AnyCancellable>()
+        let expectation = expectation(description: "wait for async process")
+
+        let subject = PassthroughSubject<Task<String, TestError>, Never>()
+
+        subject
+            .removeExpired() // Filter the 2 expired task
+            .removeDuplicates() // Pass only the first success task because the expired they never get here!
+            .sink { task in
+                XCTAssertFalse(task.isExpired)
+                expectation.fulfill()
+            }
+            .store(in: &cancellables)
+
+        // Send 2 unexpired and 2 expired:
+        subject.send(taskSuccess1)
+        subject.send(taskSuccessExpired)
+        subject.send(taskFailureExpired)
+        subject.send(taskSuccess1)
+
+        waitForExpectations(timeout: 2)
+    }
 }
diff --git a/Tests/TaskTests.swift b/Tests/TaskTests.swift
@@ -15,6 +15,7 @@ class TaskTests: XCTestCase {
         XCTAssertFalse(task.isTerminal)
         XCTAssertFalse(task.isSuccessful)
         XCTAssertFalse(task.isRecentlySucceeded)
+        XCTAssertFalse(task.isExpired)
     }
 
     func test_check_states_for_success_task() {
@@ -28,6 +29,7 @@ class TaskTests: XCTestCase {
         XCTAssertFalse(task.isFailure)
         XCTAssertTrue(task.isTerminal)
         XCTAssertTrue(task.isSuccessful)
+        XCTAssertFalse(task.isExpired)
     }
 
     func test_check_states_for_failure_task() {
@@ -42,6 +44,45 @@ class TaskTests: XCTestCase {
         XCTAssertTrue(task.isTerminal)
         XCTAssertFalse(task.isSuccessful)
         XCTAssertFalse(task.isRecentlySucceeded)
+        XCTAssertFalse(task.isExpired)
+    }
+
+    func test_check_expiration_for_custom() {
+        let expectation = expectation(description: "wait for async process")
+        expectation.expectedFulfillmentCount = 2
+
+        let task: Task<String, NSError> = .success("hola", expiration: .custom(3))
+
+        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
+            XCTAssertFalse(task.isExpired)
+            expectation.fulfill()
+        }
+
+        DispatchQueue.main.asyncAfter(deadline: .now() + 4) {
+            XCTAssertTrue(task.isExpired)
+            expectation.fulfill()
+        }
+
+        waitForExpectations(timeout: 5)
+    }
+
+    func test_check_expiration_for_immediately() {
+        let expectation = expectation(description: "wait for async process")
+        expectation.expectedFulfillmentCount = 2
+
+        let task: Task<String, NSError> = .success("hola", expiration: .immediately)
+
+        DispatchQueue.main.asyncAfter(deadline: .now()) {
+            XCTAssertFalse(task.isExpired)
+            expectation.fulfill()
+        }
+
+        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
+            XCTAssertTrue(task.isExpired)
+            expectation.fulfill()
+        }
+
+        waitForExpectations(timeout: 3)
     }
 
     func test_data_and_progress() {