Support Linux using CombineX

Package.swift

@@ -2,6 +2,16 @@
 
 import PackageDescription
 
+var dependencies: [Package.Dependency] = [
+    .package(url: "https://github.com/ReactiveX/RxSwift.git", from: "6.0.0")
+]
+var targetDependencies: [Target.Dependency] = ["RxSwift", "RxRelay"]
+
+#if os(Linux)
+dependencies.append(.package(url: "https://github.com/cx-org/CombineX", from: "0.4.0"))
+targetDependencies.append("CombineX")
+#endif
+
 let package = Package(
     name: "RxCombine",
     platforms: [
@@ -10,15 +20,14 @@ let package = Package(
     products: [
         .library(
             name: "RxCombine",
-            targets: ["RxCombine"]),
-    ],
-    dependencies: [
-        .package(url: "https://github.com/ReactiveX/RxSwift.git", from: "6.0.0")
+            targets: ["RxCombine"]
+        ),
     ],
+    dependencies: dependencies,
     targets: [
         .target(
             name: "RxCombine",
-            dependencies: ["RxSwift", "RxRelay"],
+            dependencies: targetDependencies,
             path: "Sources"),
         .testTarget(
             name: "RxCombineTests",

Sources/Combine+Rx/Publisher+Rx.swift

@@ -6,8 +6,12 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)

Sources/Combine+Rx/Subject+Rx.swift

@@ -6,8 +6,15 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+public typealias Subject = Combine.Subject
+#elseif canImport(CombineX)
+import CombineX
+public typealias Subject = CombineX.Subject
+#endif
 import RxSwift
 import RxRelay
 
@@ -17,7 +24,7 @@ import RxRelay
 /// - note: This only works when the underlying Failure is Swift.Error,
 ///         since RxSwift has no typed errors.
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-public protocol AnyObserverConvertible: Combine.Subject where Failure == Swift.Error {
+public protocol AnyObserverConvertible: Subject where Failure == Swift.Error {
     associatedtype Output
 
     /// Returns a RxSwift `AnyObserver` wrapping the Subject

Sources/Common/DemandBuffer.swift

@@ -6,8 +6,12 @@
 //  Copyright © 2020 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import class Foundation.NSRecursiveLock
 
 /// A buffer responsible for managing the demand of a downstream

Sources/Rx+Combine/BehaviorRelay+Combine.swift

@@ -5,16 +5,20 @@
 //  Created by Shai Mishali on 10/05/2020.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 import RxRelay
 
 // MARK: - Behavior Relay as Combine Subject
 
 /// A bi-directional wrapper for a RxSwift Behavior Relay
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-public final class RxCurrentValueRelay<Output>: Combine.Subject {
+public final class RxCurrentValueRelay<Output>: Subject {
     private let rxRelay: BehaviorRelay<Output>
     private let subject: CurrentValueSubject<Output, Never>
     private let subscription: AnyCancellable?

Sources/Rx+Combine/BehaviorSubject+Combine.swift

@@ -6,15 +6,19 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 
 // MARK: - Behavior Subject as Combine Subject
 
 /// A bi-directional wrapper for a RxSwift Behavior Subject
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-public final class RxCurrentValueSubject<Output>: Combine.Subject {
+public final class RxCurrentValueSubject<Output>: Subject {
     private let rxSubject: BehaviorSubject<Output>
     private let subject: CurrentValueSubject<Output, Failure>
     private let subscription: AnyCancellable?

Sources/Rx+Combine/Observable+Combine.swift

@@ -6,8 +6,12 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)

Sources/Rx+Combine/PublishRelay+Combine.swift

@@ -5,16 +5,20 @@
 //  Created by Shai Mishali on 10/05/2020.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 import RxRelay
 
 // MARK: - Publish Relay as Combine Subject
 
 /// A bi-directional wrapper for a RxSwift Publish Relay
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-public final class RxPassthroughRelay<Output>: Combine.Subject {
+public final class RxPassthroughRelay<Output>: Subject {
     private let rxRelay: PublishRelay<Output>
     private let subject = PassthroughSubject<Output, Never>()
     private let subscription: AnyCancellable?

Sources/Rx+Combine/PublishSubject+Combine.swift

@@ -6,15 +6,19 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 
 // MARK: - Behavior Subject as Combine Subject
 
 /// A bi-directional wrapper for a RxSwift Publish Subject
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-public final class RxPassthroughSubject<Output>: Combine.Subject {
+public final class RxPassthroughSubject<Output>: Subject {
     private let rxSubject: PublishSubject<Output>
     private let subject = PassthroughSubject<Output, Failure>()
     private let subscription: AnyCancellable?

Sources/Rx+Combine/Relays+Combine.swift

@@ -6,8 +6,12 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 import RxSwift
 import RxRelay
 

Sources/Rx+Combine/RxSubscription.swift

@@ -6,13 +6,20 @@
 //  Copyright © 2020 Combine Community. All rights reserved.
 //
 
+#if canImport(Combine) || canImport(CombineX)
 #if canImport(Combine)
 import Combine
+@available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
+private typealias Subscription = Combine.Subscription
+#elseif canImport(CombineX)
+import CombineX
+private typealias Subscription = CombineX.Subscription
+#endif
 import RxSwift
 
 // MARK: - Fallible
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-class RxSubscription<Upstream: ObservableConvertibleType, Downstream: Subscriber>: Combine.Subscription where Downstream.Input == Upstream.Element, Downstream.Failure == Swift.Error {
+class RxSubscription<Upstream: ObservableConvertibleType, Downstream: Subscriber>: Subscription where Downstream.Input == Upstream.Element, Downstream.Failure == Swift.Error {
     private var disposable: Disposable?
     private let buffer: DemandBuffer<Downstream>
 
@@ -52,7 +59,7 @@ extension RxSubscription: CustomStringConvertible {
 
 // MARK: - Infallible
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
-class RxInfallibleSubscription<Upstream: ObservableConvertibleType, Downstream: Subscriber>: Combine.Subscription where Downstream.Input == Upstream.Element, Downstream.Failure == Never {
+class RxInfallibleSubscription<Upstream: ObservableConvertibleType, Downstream: Subscriber>: Subscription where Downstream.Input == Upstream.Element, Downstream.Failure == Never {
     private var disposable: Disposable?
     private let buffer: DemandBuffer<Downstream>
 

Tests/ObservableAsPublisherTests.swift

@@ -6,11 +6,16 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
-#if !os(watchOS)
+#if !os(watchOS) && (canImport(Combine) || canImport(CombineX))
 import XCTest
 import RxCombine
 import RxSwift
+
+#if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
 class ObservableAsPublisherTests: XCTestCase {

Tests/PublisherAsObservableTests.swift

@@ -5,11 +5,16 @@
 //  Created by Shai Mishali on 21/03/2020.
 //
 
-#if !os(watchOS)
+#if !os(watchOS) && (canImport(Combine) || canImport(CombineX))
 import XCTest
 import RxCombine
 import RxSwift
+
+#if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
 class PublisherAsObservableTests: XCTestCase {
@@ -19,6 +24,7 @@ class PublisherAsObservableTests: XCTestCase {
         disposeBag = .init()
     }
 
+#if canImport(Combine)
     func testIntPublisher() {
         let source = (1...100).publisher
         var events = [RxSwift.Event<Int>]()
@@ -62,6 +68,51 @@ class PublisherAsObservableTests: XCTestCase {
 
         XCTAssertEqual(events, (1...14).map { .next($0) } + [.error(FakeError.ohNo)])
     }
+#elseif canImport(CombineX)
+    func testIntPublisher() {
+        let source = (1...100).cx.publisher
+        var events = [RxSwift.Event<Int>]()
+
+        source
+            .asObservable()
+            .subscribe { events.append($0) }
+            .disposed(by: disposeBag)
+
+        XCTAssertEqual(events,
+                       (1...100).map { .next($0) } + [.completed])
+    }
+
+    func testStringPublisher() {
+        let input = "Hello world I'm a RxSwift Observable".components(separatedBy: " ")
+        let source = input.cx.publisher
+        var events = [RxSwift.Event<String>]()
+
+        source
+            .asObservable()
+            .subscribe { events.append($0) }
+            .disposed(by: disposeBag)
+
+        XCTAssertEqual(events, input.map { .next($0) } + [.completed])
+    }
+
+    func testFailingPublisher() {
+        let source = (1...100).cx.publisher
+        var events = [RxSwift.Event<Int>]()
+
+        source
+            .setFailureType(to: FakeError.self)
+            .tryMap { val -> Int in
+                guard val < 15 else { throw FakeError.ohNo }
+                return val
+            }
+            .asObservable()
+            .subscribe { events.append($0) }
+            .disposed(by: disposeBag)
+
+
+        XCTAssertEqual(events, (1...14).map { .next($0) } + [.error(FakeError.ohNo)])
+    }
+#endif
 }
 
 

Tests/RxRelaysToCombineTests.swift

@@ -5,12 +5,17 @@
 //  Created by Shai Mishali on 21/03/2020.
 //
 
-#if !os(watchOS)
+#if !os(watchOS) && (canImport(Combine) || canImport(CombineX))
 import XCTest
 import RxCombine
 import RxSwift
 import RxRelay
+
+#if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
 class RxRelaysToCombineTests: XCTestCase {

Tests/RxSubjectsToCombineTests.swift

@@ -6,11 +6,16 @@
 //  Copyright © 2019 Combine Community. All rights reserved.
 //
 
-#if !os(watchOS)
+#if !os(watchOS) && (canImport(Combine) || canImport(CombineX))
 import XCTest
 import RxCombine
 import RxSwift
+
+#if canImport(Combine)
 import Combine
+#elseif canImport(CombineX)
+import CombineX
+#endif
 
 @available(OSX 10.15, iOS 13.0, tvOS 13.0, watchOS 6.0, *)
 class RxBehaviorSubjectToCombineTests: XCTestCase {