Android support

.github/workflows/ci.yml

@@ -7,6 +7,7 @@ on:
   pull_request:
     branches:
       - '*'
+  workflow_dispatch:
 
 jobs:
   library:
@@ -37,3 +38,14 @@ jobs:
       - uses: actions/checkout@v4
       - name: Run tests
         run: swift test -c ${{ matrix.config }}
+
+  android:
+    strategy:
+      matrix:
+        swift:
+          - "6.0.2"
+    name: Android
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: skiptools/swift-android-action@v2

Package.resolved

@@ -1,23 +1,33 @@
 {
+  "originHash" : "6cb737864cfe2c856014e8676e81c48a4c5377db84f4f833ddc4da453b72d94f",
   "pins" : [
+    {
+      "identity" : "opencombine",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/OpenCombine/OpenCombine.git",
+      "state" : {
+        "revision" : "8576f0d579b27020beccbccc3ea6844f3ddfc2c2",
+        "version" : "0.14.0"
+      }
+    },
     {
       "identity" : "swift-concurrency-extras",
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/pointfreeco/swift-concurrency-extras",
       "state" : {
-        "revision" : "bb5059bde9022d69ac516803f4f227d8ac967f71",
-        "version" : "1.1.0"
+        "branch" : "main",
+        "revision" : "d1e46428ceb9ffc5eb096b3ddcd89b82458500e3"
       }
     },
     {
       "identity" : "xctest-dynamic-overlay",
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/pointfreeco/xctest-dynamic-overlay",
       "state" : {
-        "revision" : "357ca1e5dd31f613a1d43320870ebc219386a495",
-        "version" : "1.2.2"
+        "revision" : "a3f634d1a409c7979cabc0a71b3f26ffa9fc8af1",
+        "version" : "1.4.3"
       }
     }
   ],
-  "version" : 2
+  "version" : 3
 }

Package.swift

@@ -19,19 +19,26 @@ let package = Package(
   dependencies: [
     .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", from: "1.0.0"),
     .package(url: "https://github.com/pointfreeco/xctest-dynamic-overlay", from: "1.2.2"),
+    .package(url: "https://github.com/OpenCombine/OpenCombine.git", from: "0.14.0")
   ],
   targets: [
     .target(
       name: "CombineSchedulers",
       dependencies: [
         .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
         .product(name: "IssueReporting", package: "xctest-dynamic-overlay"),
+        .product(name: "OpenCombineShim", package: "OpenCombine"),
+        .product(name: "OpenCombineFoundation", package: "OpenCombine"),
+        .product(name: "OpenCombineDispatch", package: "OpenCombine"),
       ]
     ),
     .testTarget(
       name: "CombineSchedulersTests",
       dependencies: [
-        "CombineSchedulers"
+        "CombineSchedulers",
+        .product(name: "OpenCombineShim", package: "OpenCombine"),
+        .product(name: "OpenCombineFoundation", package: "OpenCombine"),
+        .product(name: "OpenCombineDispatch", package: "OpenCombine")
       ]
     ),
   ]

Package@swift-6.0.swift

@@ -17,21 +17,28 @@ let package = Package(
     )
   ],
   dependencies: [
-    .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", from: "1.0.0"),
+    .package(url: "https://github.com/pointfreeco/swift-concurrency-extras", branch: "main"),
     .package(url: "https://github.com/pointfreeco/xctest-dynamic-overlay", from: "1.2.2"),
+    .package(url: "https://github.com/OpenCombine/OpenCombine.git", from: "0.14.0")
   ],
   targets: [
     .target(
       name: "CombineSchedulers",
       dependencies: [
         .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
         .product(name: "IssueReporting", package: "xctest-dynamic-overlay"),
+        .product(name: "OpenCombineShim", package: "OpenCombine"),
+        .product(name: "OpenCombineFoundation", package: "OpenCombine"),
+        .product(name: "OpenCombineDispatch", package: "OpenCombine"),
       ]
     ),
     .testTarget(
       name: "CombineSchedulersTests",
       dependencies: [
-        "CombineSchedulers"
+        "CombineSchedulers",
+        .product(name: "OpenCombineShim", package: "OpenCombine"),
+        .product(name: "OpenCombineFoundation", package: "OpenCombine"),
+        .product(name: "OpenCombineDispatch", package: "OpenCombine")
       ]
     ),
   ],

Sources/CombineSchedulers/AnyScheduler.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import Foundation
 
   /// A type-erasing wrapper for the `Scheduler` protocol, which can be useful for being generic over
@@ -237,7 +237,7 @@
   /// time type and options type.
   public typealias AnySchedulerOf<Scheduler> = AnyScheduler<
     Scheduler.SchedulerTimeType, Scheduler.SchedulerOptions
-  > where Scheduler: Combine.Scheduler
+  > where Scheduler: OpenCombineShim.Scheduler
 
   extension Scheduler {
     /// Wraps this scheduler with a type eraser.

Sources/CombineSchedulers/Concurrency.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  @preconcurrency import Combine
+#if canImport(OpenCombineShim)
+  @preconcurrency import OpenCombineShim
   import ConcurrencyExtras
 
   extension Scheduler {

Sources/CombineSchedulers/ImmediateScheduler.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import Foundation
 
   /// A scheduler for performing synchronous actions.
@@ -193,5 +193,5 @@
   /// the time type and options type.
   public typealias ImmediateSchedulerOf<Scheduler> = ImmediateScheduler<
     Scheduler.SchedulerTimeType, Scheduler.SchedulerOptions
-  > where Scheduler: Combine.Scheduler
+  > where Scheduler: OpenCombineShim.Scheduler
 #endif

Sources/CombineSchedulers/Internal/Deprecations.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import Foundation
 
   // NB: Soft-deprecated after 0.5.3:

Sources/CombineSchedulers/Internal/Lock.swift

@@ -42,4 +42,34 @@
       os_unfair_lock_unlock(self)
     }
   }
+#else
+import pthread
+struct Lock {
+  private let lockPtr: UnsafeMutablePointer<pthread_mutex_t>
+
+  internal init() {
+    lockPtr = UnsafeMutablePointer<pthread_mutex_t>.allocate(capacity: 1)
+    var attr = pthread_mutexattr_t()
+    pthread_mutexattr_settype(&attr, Int32(PTHREAD_MUTEX_RECURSIVE))
+    pthread_mutex_init(lockPtr, &attr)
+  }
+
+  internal func cleanupLock() {
+    pthread_mutex_destroy(lockPtr)
+    lockPtr.deinitialize(count: 1)
+    lockPtr.deallocate()
+  }
+
+  internal func lock() {
+    pthread_mutex_lock(lockPtr)
+  }
+
+  internal func tryLock() -> Bool {
+    return pthread_mutex_trylock(lockPtr) == 0
+  }
+
+  internal func unlock() {
+    pthread_mutex_unlock(lockPtr)
+  }
+}
 #endif

Sources/CombineSchedulers/SwiftUI.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim) && canImport(SwiftUI)
+  import OpenCombineShim
   import SwiftUI
 
   extension Scheduler {

Sources/CombineSchedulers/TestScheduler.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import ConcurrencyExtras
   import Foundation
 
@@ -283,5 +283,5 @@
   /// time type and options type.
   public typealias TestSchedulerOf<Scheduler> = TestScheduler<
     Scheduler.SchedulerTimeType, Scheduler.SchedulerOptions
-  > where Scheduler: Combine.Scheduler
+  > where Scheduler: OpenCombineShim.Scheduler
 #endif

Sources/CombineSchedulers/Timer.swift

@@ -11,12 +11,12 @@
 //===----------------------------------------------------------------------===//
 
 // Only support 64bit
-#if !(os(iOS) && (arch(i386) || arch(arm))) && canImport(Combine)
-  import Combine
+#if !(os(iOS) && (arch(i386) || arch(arm))) && canImport(OpenCombineShim)
+  import OpenCombineShim
   import Foundation
 
   @available(macOS 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-  extension Scheduler {
+  extension OpenCombineShim.Scheduler {
     /// Returns a publisher that repeatedly emits the scheduler's current time on the given
     /// interval.
     ///
@@ -80,7 +80,7 @@
     /// scheduler.advance(by: 1_000)
     /// XCTAssertEqual(output, Array(0...1_001))
     /// ```
-    public final class Timer<Scheduler: Combine.Scheduler>: ConnectablePublisher {
+    public final class Timer<Scheduler: OpenCombineShim.Scheduler>: ConnectablePublisher {
       public typealias Output = Scheduler.SchedulerTimeType
       public typealias Failure = Never
 
@@ -330,6 +330,33 @@
           demand += n
         }
 
+        #if os(Android)
+        func timerFired() {
+          lock.lock()
+          guard let ds = downstream, let parent = self.parent else {
+            lock.unlock()
+            return
+          }
+
+          // This publisher drops events on the floor when there is no space in the subscriber
+          guard demand > 0 else {
+            lock.unlock()
+            return
+          }
+
+          demand -= 1
+          lock.unlock()
+
+          let extra = ds.receive(parent.scheduler.now)
+          guard extra > 0 else {
+            return
+          }
+
+          lock.lock()
+          demand += extra
+          lock.unlock()
+        }
+        #else
         @objc
         func timerFired() {
           lock.lock()
@@ -356,6 +383,7 @@
           demand += extra
           lock.unlock()
         }
+        #endif
       }
     }
   }

Sources/CombineSchedulers/UIScheduler.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
 
   #if swift(>=6)
     @preconcurrency import Dispatch

Sources/CombineSchedulers/UnimplementedScheduler.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import Foundation
   import IssueReporting
 
@@ -267,5 +267,5 @@
   /// by the time type and options type.
   public typealias UnimplementedSchedulerOf<Scheduler> = UnimplementedScheduler<
     Scheduler.SchedulerTimeType, Scheduler.SchedulerOptions
-  > where Scheduler: Combine.Scheduler
+  > where Scheduler: OpenCombineShim.Scheduler
 #endif

Tests/CombineSchedulersTests/ImmediateSchedulerTests.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import CombineSchedulers
   import XCTest
 

Tests/CombineSchedulersTests/TestSchedulerTests.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import CombineSchedulers
   import ConcurrencyExtras
   import XCTest
@@ -176,6 +176,7 @@
       XCTAssertEqual(values, [1, 42, 42, 1, 42])
     }
 
+    #if !os(Android)
     func testAdvanceToFarFuture() async {
       await withMainSerialExecutor {
         var cancellables: Set<AnyCancellable> = []
@@ -262,7 +263,8 @@
         XCTAssertEqual(count, 10)
       }
     }
-
+    #endif
+
     func testNowIsAdvanced() {
       let testScheduler = DispatchQueue.test
       let start = testScheduler.now

Tests/CombineSchedulersTests/TimerTests.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import CombineSchedulers
   import XCTest
 
@@ -77,6 +77,7 @@
       )
     }
 
+    #if !os(Android)
     func testInterleavingTimers() {
       let scheduler = DispatchQueue.test
 
@@ -106,7 +107,8 @@
       scheduler.advance(by: 1)
       XCTAssertEqual(output, [1, 2, 1, 1, 2])
     }
-
+    #endif
+
     func testTimerCancellation() {
       let scheduler = DispatchQueue.test
 

Tests/CombineSchedulersTests/UISchedulerTests.swift

@@ -1,5 +1,5 @@
-#if canImport(Combine)
-  import Combine
+#if canImport(OpenCombineShim)
+  import OpenCombineShim
   import CombineSchedulers
   import ConcurrencyExtras
   import XCTest