Repository: broadwaylamb/OpenCombine
Pull request #123 (2019-12-11 16:44:55)

Update for Xcode 11.3
- Send subscription synchronously in `ReceiveOn` and `Delay` operators
- Some locks made recursive, as they should be
- `ObservableObjectPublisher` doesn't use `PassthroughSubject` under the hood anymore

codecov[bot]: # [Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=h1) Report
> Merging [#123](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=desc) into [master](https://codecov.io/gh/broadwaylamb/OpenCombine/commit/95b42abce3ef1f22e8cc984beb441c63577b156e?src=pr&el=desc) will **decrease** coverage by `0.29%`.
> The diff coverage is `96.55%`.

[![Impacted file tree graph](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/graphs/tree.svg?width=650&token=XMi8wfbUeP&height=150&src=pr)](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master     #123     +/-   ##
=========================================
- Coverage   98.12%   97.82%   -0.3%     
=========================================
  Files          76       75      -1     
  Lines        4003     4009      +6     
=========================================
- Hits         3928     3922      -6     
- Misses         75       87     +12
```


| [Impacted Files](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [.../OpenCombine/Publishers/Publishers.DropWhile.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuRHJvcFdoaWxlLnN3aWZ0) | `100% <ø> (ø)` | :arrow_up: |
| [...rces/OpenCombine/Publishers/Publishers.Print.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuUHJpbnQuc3dpZnQ=) | `100% <ø> (ø)` | :arrow_up: |
| [...enCombine/Publishers/Publishers.HandleEvents.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuSGFuZGxlRXZlbnRzLnN3aWZ0) | `100% <ø> (ø)` | :arrow_up: |
| [...penCombine/Publishers/Publishers.Autoconnect.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuQXV0b2Nvbm5lY3Quc3dpZnQ=) | `100% <ø> (ø)` | :arrow_up: |
| [...urces/OpenCombine/Publishers/Publishers.Drop.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuRHJvcC5zd2lmdA==) | `100% <ø> (ø)` | :arrow_up: |
| [...es/OpenCombine/Publishers/Publishers.FlatMap.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuRmxhdE1hcC5zd2lmdA==) | `100% <ø> (ø)` | :arrow_up: |
| [...ces/OpenCombine/Publishers/Publishers.Output.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuT3V0cHV0LnN3aWZ0) | `100% <ø> (ø)` | :arrow_up: |
| [...penCombine/Publishers/Publishers.SubscribeOn.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuU3Vic2NyaWJlT24uc3dpZnQ=) | `100% <ø> (ø)` | :arrow_up: |
| [Sources/OpenCombine/Publishers/Record.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1JlY29yZC5zd2lmdA==) | `100% <ø> (ø)` | :arrow_up: |
| [...penCombine/Publishers/Publishers.Concatenate.swift](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree#diff-U291cmNlcy9PcGVuQ29tYmluZS9QdWJsaXNoZXJzL1B1Ymxpc2hlcnMuQ29uY2F0ZW5hdGUuc3dpZnQ=) | `100% <ø> (ø)` | :arrow_up: |
| ... and [20 more](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=footer). Last update [95b42ab...d24e20c](https://codecov.io/gh/broadwaylamb/OpenCombine/pull/123?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

broadwaylamb: Let's wait until CircleCI supports Xcode 11.3.

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 6fce731..96889c0 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -1,10 +1,10 @@
 version: 2
 jobs:
-  "Execute tests on macOS 10.15.0 (Xcode 11.2.0, Swift 5.1.2)":
+  "Execute tests on macOS 10.15.0 (Xcode 11.3.0, Swift 5.1.3)":
     macos:
-      xcode: "11.2.0"
+      xcode: "11.3.0"
     environment:
-      SWIFT_VERSION: "5.1.2"
+      SWIFT_VERSION: "5.1.3"
     steps:
       - checkout
       - run:
@@ -60,35 +60,35 @@ jobs:
           command: |
             bash <(curl -s https://codecov.io/bash) -D DerivedData
 
-  "Execute compatibility tests on iOS 13.2.2 (Xcode 11.2.0, Swift 5.1.2)":
+  "Execute compatibility tests on iOS 13.3 (Xcode 11.3.0, Swift 5.1.3)":
     macos:
-      xcode: "11.2.0"
+      xcode: "11.3.0"
     environment:
-      SWIFT_VERSION: "5.1.2"
+      SWIFT_VERSION: "5.1.3"
     steps:
       - checkout
       - run:
           name: Generating Xcode project
           command: make generate-compatibility-xcodeproj
       - run:
-          name: Building for testing on iOS 13.2.2 with xcodebuild
+          name: Building for testing on iOS 13.3 with xcodebuild
           command: |
             set -o pipefail \
               && xcodebuild build-for-testing \
                    -scheme OpenCombine-Package \
-                   -destination "platform=iOS Simulator,name=iPhone 11,OS=13.2.2" \
+                   -destination "platform=iOS Simulator,name=iPhone 11,OS=13.3" \
                    -derivedDataPath DerivedData \
               | tee xcodebuild_build-for-testing.log \
               | xcpretty
       - store_artifacts:
           path: xcodebuild_build-for-testing.log
       - run:
-          name: Testing against Combine on iOS 13.2.2 with xcodebuild
+          name: Testing against Combine on iOS 13.3 with xcodebuild
           command: |
             set -o pipefail \
               && xcodebuild test-without-building \
                    -scheme OpenCombine-Package \
-                   -destination "platform=iOS Simulator,name=iPhone 11,OS=13.2.2" \
+                   -destination "platform=iOS Simulator,name=iPhone 11,OS=13.3" \
                    -derivedDataPath DerivedData \
               | tee xcodebuild_test-without-building.log \
               | xcpretty --report junit -o build/reports/results.xml
@@ -217,7 +217,7 @@ jobs:
 
   "Run SwiftLint and Danger":
     macos:
-      xcode: "11.2.0"
+      xcode: "11.3.0"
     environment:
       HOMEBREW_NO_AUTO_UPDATE: "1"
     steps:
@@ -236,7 +236,7 @@ jobs:
 
   "Run Pod spec lint":
     macos:
-      xcode: "11.2.0"
+      xcode: "11.3.0"
     environment:
       HOMEBREW_NO_AUTO_UPDATE: "1"
     steps:
@@ -250,10 +250,10 @@ workflows:
   version: 2
   "OpenCombine: execute tests on macOS":
     jobs:
-      - "Execute tests on macOS 10.15.0 (Xcode 11.2.0, Swift 5.1.2)"
+      - "Execute tests on macOS 10.15.0 (Xcode 11.3.0, Swift 5.1.3)"
   "OpenCombine: execute compatibility tests":
     jobs:
-      - "Execute compatibility tests on iOS 13.2.2 (Xcode 11.2.0, Swift 5.1.2)"
+      - "Execute compatibility tests on iOS 13.3 (Xcode 11.3.0, Swift 5.1.3)"
   "OpenCombine: execute tests on iOS":
     jobs:
       - "Execute tests on iOS 9.3 (Xcode 10.2.1, Swift 5.0.1)"
diff --git a/OpenCombineFoundation.podspec b/OpenCombineFoundation.podspec
new file mode 100644
index 0000000..3f9dd26
--- /dev/null
+++ b/OpenCombineFoundation.podspec
@@ -0,0 +1,25 @@
+Pod::Spec.new do |spec|
+    spec.name          = "OpenCombineFoundation"
+    spec.version       = "0.7.0"
+    spec.summary       = "OpenCombine + OpenCombineFoundation interoperability"
+  
+    spec.description   = <<-DESC
+    Adds publishers to Foundation types like NotificationCenter, URLSession etc.
+    DESC
+  
+    spec.homepage      = "https://github.com/broadwaylamb/OpenCombine/"
+    spec.license       = "MIT"
+  
+    spec.authors       = { "Sergej Jaskiewicz" => "jaskiewiczs@icloud.com" }
+    spec.source        = { :git => "https://github.com/broadwaylamb/OpenCombine.git", :tag => "#{spec.version}" }
+  
+    spec.swift_version = "5.0"
+  
+    spec.osx.deployment_target     = "10.10"
+    spec.ios.deployment_target     = "8.0"
+    spec.watchos.deployment_target = "2.0"
+    spec.tvos.deployment_target    = "9.0"
+  
+    spec.source_files = "Sources/OpenCombineFoundation/**/*.swift"
+    spec.dependency     "OpenCombine", '~> 0.6'
+  end
\ No newline at end of file
diff --git a/Package.swift b/Package.swift
index 1158d57..20fd3f6 100644
--- a/Package.swift
+++ b/Package.swift
@@ -7,14 +7,18 @@ let package = Package(
     products: [
         .library(name: "OpenCombine", targets: ["OpenCombine"]),
         .library(name: "OpenCombineDispatch", targets: ["OpenCombineDispatch"]),
+        .library(name: "OpenCombineFoundation", targets: ["OpenCombineFoundation"]),
     ],
     targets: [
         .target(name: "COpenCombineHelpers"),
         .target(name: "OpenCombine", dependencies: ["COpenCombineHelpers"]),
         .target(name: "OpenCombineDispatch", dependencies: ["OpenCombine"]),
+        .target(name: "OpenCombineFoundation", dependencies: ["OpenCombine",
+                                                              "COpenCombineHelpers"]),
         .testTarget(name: "OpenCombineTests",
                     dependencies: ["OpenCombine",
-                                   "OpenCombineDispatch"],
+                                   "OpenCombineDispatch",
+                                   "OpenCombineFoundation"],
                     swiftSettings: [.unsafeFlags(["-enable-testing"])])
     ],
     cxxLanguageStandard: .cxx1z
diff --git a/RemainingFoundationInterface.swift b/RemainingFoundationInterface.swift
new file mode 100644
index 0000000..0f2416d
--- /dev/null
+++ b/RemainingFoundationInterface.swift
@@ -0,0 +1,163 @@
+// From /Applications/Xcode.app/Contents/Developer/Platforms/MacOSX.platform/Developer/SDKs/MacOSX.sdk/usr/lib/swift/Foundation.swiftmodule/x86_64.swiftinterface
+// swift-interface-format-version: 1.0
+// swift-compiler-version: Apple Swift version 5.1.1 (swiftlang-1100.8.275.1 clang-1100.0.32.1)
+// swift-module-flags: -target x86_64-apple-macosx10.15 -enable-objc-interop -autolink-force-load -enable-library-evolution -module-link-name swiftFoundation -swift-version 5 -O -enforce-exclusivity=unchecked -module-name Foundation
+
+public typealias Published = Combine.Published
+
+public typealias ObservableObject = Combine.ObservableObject
+
+public protocol _KeyValueCodingAndObservingPublishing {
+}
+
+extension NSObject : Foundation._KeyValueCodingAndObservingPublishing {
+}
+
+extension _KeyValueCodingAndObservingPublishing where Self : ObjectiveC.NSObject {
+  public func publisher<Value>(for keyPath: Swift.KeyPath<Self, Value>, options: Foundation.NSKeyValueObservingOptions = [.initial, .new]) -> ObjectiveC.NSObject.KeyValueObservingPublisher<Self, Value>
+}
+
+extension NSObject.KeyValueObservingPublisher {
+  public func didChange() -> Combine.Publishers.Map<ObjectiveC.NSObject.KeyValueObservingPublisher<Subject, Value>, Swift.Void>
+}
+
+extension NSObject {
+  public struct KeyValueObservingPublisher<Subject, Value> : Swift.Equatable where Subject : ObjectiveC.NSObject {
+    public let object: Subject
+    public let keyPath: Swift.KeyPath<Subject, Value>
+    public let options: Foundation.NSKeyValueObservingOptions
+    public init(object: Subject, keyPath: Swift.KeyPath<Subject, Value>, options: Foundation.NSKeyValueObservingOptions)
+    public static func == (lhs: ObjectiveC.NSObject.KeyValueObservingPublisher<Subject, Value>, rhs: ObjectiveC.NSObject.KeyValueObservingPublisher<Subject, Value>) -> Swift.Bool
+  }
+}
+
+extension NSObject.KeyValueObservingPublisher : Combine.Publisher {
+  public typealias Output = Value
+  public typealias Failure = Swift.Never
+  public func receive<S>(subscriber: S) where Value == S.Input, S : Combine.Subscriber, S.Failure == ObjectiveC.NSObject.KeyValueObservingPublisher<Subject, Value>.Failure
+}
+
+extension Timer {
+  public static func publish(every interval: Foundation.TimeInterval, tolerance: Foundation.TimeInterval? = nil, on runLoop: Foundation.RunLoop, in mode: Foundation.RunLoop.Mode, options: Foundation.RunLoop.SchedulerOptions? = nil) -> Foundation.Timer.TimerPublisher
+  final public class TimerPublisher : Combine.ConnectablePublisher {
+    public typealias Output = Foundation.Date
+    public typealias Failure = Swift.Never
+    final public let interval: Foundation.TimeInterval
+    final public let tolerance: Foundation.TimeInterval?
+    final public let runLoop: Foundation.RunLoop
+    final public let mode: Foundation.RunLoop.Mode
+    final public let options: Foundation.RunLoop.SchedulerOptions?
+    public init(interval: Foundation.TimeInterval, tolerance: Foundation.TimeInterval? = nil, runLoop: Foundation.RunLoop, mode: Foundation.RunLoop.Mode, options: Foundation.RunLoop.SchedulerOptions? = nil)
+    final public func receive<S>(subscriber: S) where S : Combine.Subscriber, S.Failure == Foundation.Timer.TimerPublisher.Failure, S.Input == Foundation.Timer.TimerPublisher.Output
+    final public func connect() -> Combine.Cancellable
+    @objc deinit
+  }
+}
+
+extension OperationQueue : Combine.Scheduler {
+  public struct SchedulerTimeType : Swift.Strideable, Swift.Codable, Swift.Hashable {
+    public var date: Foundation.Date
+    public init(_ date: Foundation.Date)
+    public func distance(to other: Foundation.OperationQueue.SchedulerTimeType) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+    public func advanced(by n: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Foundation.OperationQueue.SchedulerTimeType
+    public struct Stride : Swift.ExpressibleByFloatLiteral, Swift.Comparable, Swift.SignedNumeric, Swift.Codable, Combine.SchedulerTimeIntervalConvertible {
+      public typealias FloatLiteralType = Foundation.TimeInterval
+      public typealias IntegerLiteralType = Foundation.TimeInterval
+      public typealias Magnitude = Foundation.TimeInterval
+      public var magnitude: Foundation.TimeInterval
+      public var timeInterval: Foundation.TimeInterval {
+        get
+      }
+      public init(integerLiteral value: Foundation.TimeInterval)
+      public init(floatLiteral value: Foundation.TimeInterval)
+      public init(_ timeInterval: Foundation.TimeInterval)
+      public init?<T>(exactly source: T) where T : Swift.BinaryInteger
+      public static func < (lhs: Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Swift.Bool
+      public static func * (lhs: Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func + (lhs: Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func - (lhs: Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func *= (lhs: inout Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride)
+      public static func += (lhs: inout Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride)
+      public static func -= (lhs: inout Foundation.OperationQueue.SchedulerTimeType.Stride, rhs: Foundation.OperationQueue.SchedulerTimeType.Stride)
+      public static func seconds(_ s: Swift.Int) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func seconds(_ s: Swift.Double) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func milliseconds(_ ms: Swift.Int) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func microseconds(_ us: Swift.Int) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public static func nanoseconds(_ ns: Swift.Int) -> Foundation.OperationQueue.SchedulerTimeType.Stride
+      public init(from decoder: Swift.Decoder) throws
+      public func encode(to encoder: Swift.Encoder) throws
+      public static func == (a: Foundation.OperationQueue.SchedulerTimeType.Stride, b: Foundation.OperationQueue.SchedulerTimeType.Stride) -> Swift.Bool
+    }
+    public init(from decoder: Swift.Decoder) throws
+    public func encode(to encoder: Swift.Encoder) throws
+    public var hashValue: Swift.Int {
+      get
+    }
+    public func hash(into hasher: inout Swift.Hasher)
+  }
+  public struct SchedulerOptions {
+  }
+  public func schedule(options: Foundation.OperationQueue.SchedulerOptions?, _ action: @escaping () -> Swift.Void)
+  public func schedule(after date: Foundation.OperationQueue.SchedulerTimeType, tolerance: Foundation.OperationQueue.SchedulerTimeType.Stride, options: Foundation.OperationQueue.SchedulerOptions?, _ action: @escaping () -> Swift.Void)
+  public func schedule(after date: Foundation.OperationQueue.SchedulerTimeType, interval: Foundation.OperationQueue.SchedulerTimeType.Stride, tolerance: Foundation.OperationQueue.SchedulerTimeType.Stride, options: Foundation.OperationQueue.SchedulerOptions?, _ action: @escaping () -> Swift.Void) -> Combine.Cancellable
+  public var now: Foundation.OperationQueue.SchedulerTimeType {
+    get
+  }
+  public var minimumTolerance: Foundation.OperationQueue.SchedulerTimeType.Stride {
+    get
+  }
+}
+
+extension RunLoop : Combine.Scheduler {
+  public struct SchedulerTimeType : Swift.Strideable, Swift.Codable, Swift.Hashable {
+    public var date: Foundation.Date
+    public init(_ date: Foundation.Date)
+    public func distance(to other: Foundation.RunLoop.SchedulerTimeType) -> Foundation.RunLoop.SchedulerTimeType.Stride
+    public func advanced(by n: Foundation.RunLoop.SchedulerTimeType.Stride) -> Foundation.RunLoop.SchedulerTimeType
+    public struct Stride : Swift.ExpressibleByFloatLiteral, Swift.Comparable, Swift.SignedNumeric, Swift.Codable, Combine.SchedulerTimeIntervalConvertible {
+      public typealias FloatLiteralType = Foundation.TimeInterval
+      public typealias IntegerLiteralType = Foundation.TimeInterval
+      public typealias Magnitude = Foundation.TimeInterval
+      public var magnitude: Foundation.TimeInterval
+      public var timeInterval: Foundation.TimeInterval {
+        get
+      }
+      public init(integerLiteral value: Foundation.TimeInterval)
+      public init(floatLiteral value: Foundation.TimeInterval)
+      public init(_ timeInterval: Foundation.TimeInterval)
+      public init?<T>(exactly source: T) where T : Swift.BinaryInteger
+      public static func < (lhs: Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride) -> Swift.Bool
+      public static func * (lhs: Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func + (lhs: Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func - (lhs: Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func *= (lhs: inout Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride)
+      public static func += (lhs: inout Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride)
+      public static func -= (lhs: inout Foundation.RunLoop.SchedulerTimeType.Stride, rhs: Foundation.RunLoop.SchedulerTimeType.Stride)
+      public static func seconds(_ s: Swift.Int) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func seconds(_ s: Swift.Double) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func milliseconds(_ ms: Swift.Int) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func microseconds(_ us: Swift.Int) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public static func nanoseconds(_ ns: Swift.Int) -> Foundation.RunLoop.SchedulerTimeType.Stride
+      public init(from decoder: Swift.Decoder) throws
+      public func encode(to encoder: Swift.Encoder) throws
+      public static func == (a: Foundation.RunLoop.SchedulerTimeType.Stride, b: Foundation.RunLoop.SchedulerTimeType.Stride) -> Swift.Bool
+    }
+    public init(from decoder: Swift.Decoder) throws
+    public func encode(to encoder: Swift.Encoder) throws
+    public var hashValue: Swift.Int {
+      get
+    }
+    public func hash(into hasher: inout Swift.Hasher)
+  }
+  public struct SchedulerOptions {
+  }
+  public func schedule(options: Foundation.RunLoop.SchedulerOptions?, _ action: @escaping () -> Swift.Void)
+  public func schedule(after date: Foundation.RunLoop.SchedulerTimeType, tolerance: Foundation.RunLoop.SchedulerTimeType.Stride, options: Foundation.RunLoop.SchedulerOptions?, _ action: @escaping () -> Swift.Void)
+  public func schedule(after date: Foundation.RunLoop.SchedulerTimeType, interval: Foundation.RunLoop.SchedulerTimeType.Stride, tolerance: Foundation.RunLoop.SchedulerTimeType.Stride, options: Foundation.RunLoop.SchedulerOptions?, _ action: @escaping () -> Swift.Void) -> Combine.Cancellable
+  public var now: Foundation.RunLoop.SchedulerTimeType {
+    get
+  }
+  public var minimumTolerance: Foundation.RunLoop.SchedulerTimeType.Stride {
+    get
+  }
+}
diff --git a/Sources/OpenCombine/CurrentValueSubject.swift b/Sources/OpenCombine/CurrentValueSubject.swift
index 87dca6e..58692e6 100644
--- a/Sources/OpenCombine/CurrentValueSubject.swift
+++ b/Sources/OpenCombine/CurrentValueSubject.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 11.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A subject that wraps a single value and publishes a new element whenever the value
 /// changes.
 public final class CurrentValueSubject<Output, Failure: Error>: Subject {
diff --git a/Sources/OpenCombine/Future.swift b/Sources/OpenCombine/Future.swift
index 18ee64e..0f54a26 100644
--- a/Sources/OpenCombine/Future.swift
+++ b/Sources/OpenCombine/Future.swift
@@ -5,10 +5,6 @@
 //  Created by Max Desiatov on 24/11/2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A publisher that eventually produces one value and then finishes or fails.
 public final class Future<Output, Failure>: Publisher where Failure: Error {
 
diff --git a/Sources/OpenCombine/Helpers/FilterProducer.swift b/Sources/OpenCombine/Helpers/FilterProducer.swift
index 2e4d20d..226938b 100644
--- a/Sources/OpenCombine/Helpers/FilterProducer.swift
+++ b/Sources/OpenCombine/Helpers/FilterProducer.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 23.10.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A helper class that acts like both subscriber and subscription.
 ///
 /// Filter-like operators send an instance of their `Inner` class that is subclass
diff --git a/Sources/OpenCombine/Helpers/ReduceProducer.swift b/Sources/OpenCombine/Helpers/ReduceProducer.swift
index f86e728..1a6a528 100644
--- a/Sources/OpenCombine/Helpers/ReduceProducer.swift
+++ b/Sources/OpenCombine/Helpers/ReduceProducer.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 22.09.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A helper class that acts like both subscriber and subscription.
 ///
 /// Reduce-like operators send an instance of their `Inner` class that is subclass
diff --git a/Sources/OpenCombine/Helpers/SubjectSubscriber.swift b/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
index 61b2054..e95fae5 100644
--- a/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
+++ b/Sources/OpenCombine/Helpers/SubjectSubscriber.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 16/09/2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 // NOTE: This class has been audited for thread safety.
 internal final class SubjectSubscriber<Downstream: Subject>
     : Subscriber,
diff --git a/Sources/OpenCombine/ObservableObject.swift b/Sources/OpenCombine/ObservableObject.swift
index d8133a4..d449f0a 100644
--- a/Sources/OpenCombine/ObservableObject.swift
+++ b/Sources/OpenCombine/ObservableObject.swift
@@ -67,19 +67,126 @@ public final class ObservableObjectPublisher: Publisher {
 
     public typealias Failure = Never
 
-    private let subject: PassthroughSubject<Void, Never>
+    private let lock = UnfairLock.allocate()
 
-    public init() {
-        subject = .init()
+    private var connections = Set<Conduit>()
+
+    // TODO: Combine needs this for some reason
+    private var identifier: ObjectIdentifier?
+
+    public init() {}
+
+    deinit {
+        lock.deallocate()
     }
 
     public func receive<Downstream: Subscriber>(subscriber: Downstream)
         where Downstream.Input == Void, Downstream.Failure == Never
     {
-        subject.subscribe(subscriber)
+        let inner = Inner(downstream: subscriber, parent: self)
+        lock.lock()
+        connections.insert(inner)
+        lock.unlock()
+        subscriber.receive(subscription: inner)
     }
 
     public func send() {
-        subject.send()
+        lock.lock()
+        let connections = self.connections
+        lock.unlock()
+        for connection in connections {
+            connection.send()
+        }
+    }
+
+    private func remove(_ conduit: Conduit) {
+        lock.lock()
+        connections.remove(conduit)
+        lock.unlock()
+    }
+}
+
+extension ObservableObjectPublisher {
+    private class Conduit: Hashable {
+
+        fileprivate func send() {
+            abstractMethod()
+        }
+
+        fileprivate static func == (lhs: Conduit, rhs: Conduit) -> Bool {
+            return lhs === rhs
+        }
+
+        fileprivate func hash(into hasher: inout Hasher) {
+            hasher.combine(ObjectIdentifier(self))
+        }
+    }
+
+    private final class Inner<Downstream: Subscriber>
+        : Conduit,
+          Subscription,
+          CustomStringConvertible,
+          CustomReflectable,
+          CustomPlaygroundDisplayConvertible
+        where Downstream.Input == Void, Downstream.Failure == Never
+    {
+        private enum State {
+            case initialized
+            case active
+            case terminal
+        }
+
+        private weak var parent: ObservableObjectPublisher?
+        private let downstream: Downstream
+        private let downstreamLock = UnfairRecursiveLock.allocate()
+        private let lock = UnfairLock.allocate()
+        private var state = State.initialized
+
+        init(downstream: Downstream, parent: ObservableObjectPublisher) {
+            self.parent = parent
+            self.downstream = downstream
+        }
+
+        deinit {
+            downstreamLock.deallocate()
+            lock.deallocate()
+        }
+
+        override func send() {
+            lock.lock()
+            let state = self.state
+            lock.unlock()
+            if state == .active {
+                downstreamLock.lock()
+                _ = downstream.receive()
+                downstreamLock.unlock()
+            }
+        }
+
+        func request(_ demand: Subscribers.Demand) {
+            lock.lock()
+            if state == .initialized {
+                state = .active
+            }
+            lock.unlock()
+        }
+
+        func cancel() {
+            lock.lock()
+            state = .terminal
+            lock.unlock()
+            parent?.remove(self)
+        }
+
+        var description: String { return "ObservableObjectPublisher" }
+
+        var customMirror: Mirror {
+            let children = CollectionOfOne<Mirror.Child>(("downstream", downstream))
+            return Mirror(self, children: children)
+        }
+
+        var playgroundDescription: Any {
+            return description
+        }
     }
 }
diff --git a/Sources/OpenCombine/PassthroughSubject.swift b/Sources/OpenCombine/PassthroughSubject.swift
index 49892e8..63ee24f 100644
--- a/Sources/OpenCombine/PassthroughSubject.swift
+++ b/Sources/OpenCombine/PassthroughSubject.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 11.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A subject that passes along values and completion.
 ///
 /// Use a `PassthroughSubject` in unit tests when you want a publisher than can publish
diff --git a/Sources/OpenCombine/Publishers/Publishers.Autoconnect.swift b/Sources/OpenCombine/Publishers/Publishers.Autoconnect.swift
index f60741b..72f48ec 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Autoconnect.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Autoconnect.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 18/09/2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension ConnectablePublisher {
 
     /// Automates the process of connecting or disconnecting from this connectable
diff --git a/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift b/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
index c0d2c7e..4b85d2e 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Concatenate.swift
@@ -145,9 +145,7 @@ extension Publishers.Concatenate {
 
         private let lock = UnfairLock.allocate()
 
-        // ??? This lock is non-recursive in Combine, but it should be!
-        // (FB7404824 if Apple folks are watching)
-        private let downstreamLock = UnfairLock.allocate()
+        private let downstreamLock = UnfairRecursiveLock.allocate()
 
         fileprivate init(downstream: Downstream, suffix: Suffix) {
             self.downstream = downstream
diff --git a/Sources/OpenCombine/Publishers/Publishers.Delay.swift b/Sources/OpenCombine/Publishers/Publishers.Delay.swift
index 5a6f100..5386e75 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Delay.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Delay.swift
@@ -5,10 +5,6 @@
 //  Created by Евгений Богомолов on 07/09/2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Delays delivery of all output to the downstream receiver by a specified amount
@@ -104,7 +100,7 @@ extension Publishers.Delay {
 
         private let lock = UnfairLock.allocate()
         private var state: State
-        private let downstreamLock = UnfairLock.allocate()
+        private let downstreamLock = UnfairRecursiveLock.allocate()
 
         fileprivate init(_ publisher: Delay, downstream: Downstream) {
             state = .ready(publisher, downstream)
@@ -115,13 +111,7 @@ extension Publishers.Delay {
             downstreamLock.deallocate()
         }
 
-        private func schedule(_ delay: Delay,
-                              immediate: Bool,
-                              work: @escaping () -> Void) {
-            if immediate {
-                delay.scheduler.schedule(options: delay.options, work)
-                return
-            }
+        private func schedule(_ delay: Delay, work: @escaping () -> Void) {
             delay
                 .scheduler
                 .schedule(after: delay.scheduler.now.advanced(by: delay.interval),
@@ -139,18 +129,6 @@ extension Publishers.Delay {
             }
             state = .subscribed(delay, downstream, subscription)
             lock.unlock()
-            schedule(delay, immediate: true) { [weak self] in
-                self?.scheduledReceive(subscription: subscription)
-            }
-        }
-
-        private func scheduledReceive(subscription: Subscription) {
-            lock.lock()
-            guard case let .subscribed(_, downstream, _) = state else {
-                lock.unlock()
-                return
-            }
-            lock.unlock()
             downstreamLock.lock()
             downstream.receive(subscription: self)
             downstreamLock.unlock()
@@ -163,8 +141,8 @@ extension Publishers.Delay {
                 return .none
             }
             lock.unlock()
-            schedule(delay, immediate: false) { [weak self] in
-                self?.scheduledReceive(input, downstream: downstream)
+            schedule(delay) {
+                self.scheduledReceive(input, downstream: downstream)
             }
             return .none
         }
@@ -193,8 +171,8 @@ extension Publishers.Delay {
             }
             state = .terminal
             lock.unlock()
-            schedule(delay, immediate: false) { [weak self] in
-                self?.scheduledReceive(completion: completion, downstream: downstream)
+            schedule(delay) {
+                self.scheduledReceive(completion: completion, downstream: downstream)
             }
         }
 
diff --git a/Sources/OpenCombine/Publishers/Publishers.Drop.swift b/Sources/OpenCombine/Publishers/Publishers.Drop.swift
index f0db237..fb168ed 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Drop.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Drop.swift
@@ -5,10 +5,6 @@
 //  Created by Sven Weidauer on 03.10.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
     /// Omits the specified number of elements before republishing subsequent elements.
     ///
diff --git a/Sources/OpenCombine/Publishers/Publishers.DropWhile.swift b/Sources/OpenCombine/Publishers/Publishers.DropWhile.swift
index 362b8c7..d215094 100644
--- a/Sources/OpenCombine/Publishers/Publishers.DropWhile.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.DropWhile.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 16.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Omits elements from the upstream publisher until a given closure returns false,
diff --git a/Sources/OpenCombine/Publishers/Publishers.FlatMap.swift b/Sources/OpenCombine/Publishers/Publishers.FlatMap.swift
index 2288614..cbbe78f 100644
--- a/Sources/OpenCombine/Publishers/Publishers.FlatMap.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.FlatMap.swift
@@ -4,10 +4,6 @@
 //  Created by Eric Patey on 16.08.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
     /// Transforms all elements from an upstream publisher into a new or existing
     /// publisher.
@@ -92,10 +88,9 @@ extension Publishers.FlatMap {
         /// by the `downstreamLock`.
         private let lock = UnfairLock.allocate()
 
-        // Must be recursive lock. Probably a bug in Combine.
         /// All the calls to the downstream subscriber should be made with this lock
         /// acquired.
-        private let downstreamLock = UnfairLock.allocate()
+        private let downstreamLock = UnfairRecursiveLock.allocate()
 
         private let downstream: Downstream
 
diff --git a/Sources/OpenCombine/Publishers/Publishers.HandleEvents.swift b/Sources/OpenCombine/Publishers/Publishers.HandleEvents.swift
index 29b62e9..58bb2bc 100644
--- a/Sources/OpenCombine/Publishers/Publishers.HandleEvents.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.HandleEvents.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 03.12.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Performs the specified closures when publisher events occur.
diff --git a/Sources/OpenCombine/Publishers/Publishers.IgnoreOutput.swift b/Sources/OpenCombine/Publishers/Publishers.IgnoreOutput.swift
index 8fce777..1f5f2b3 100644
--- a/Sources/OpenCombine/Publishers/Publishers.IgnoreOutput.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.IgnoreOutput.swift
@@ -4,10 +4,6 @@
 //  Created by Eric Patey on 16.08.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Ingores all upstream elements, but passes along a completion
diff --git a/Sources/OpenCombine/Publishers/Publishers.Map.swift b/Sources/OpenCombine/Publishers/Publishers.Map.swift
index e1bca80..a3ce569 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Map.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Map.swift
@@ -5,10 +5,6 @@
 //  Created by Anton Nazarov on 25.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Transforms all elements from the upstream publisher with a provided closure.
diff --git a/Sources/OpenCombine/Publishers/Publishers.MeasureInterval.swift b/Sources/OpenCombine/Publishers/Publishers.MeasureInterval.swift
index 85a958b..54d32dd 100644
--- a/Sources/OpenCombine/Publishers/Publishers.MeasureInterval.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.MeasureInterval.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 03.12.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Measures and emits the time interval between events received from an upstream
diff --git a/Sources/OpenCombine/Publishers/Publishers.Multicast.swift b/Sources/OpenCombine/Publishers/Publishers.Multicast.swift
index c6a3197..e54b33e 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Multicast.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Multicast.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 14.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Applies a closure to create a subject that delivers elements to subscribers.
diff --git a/Sources/OpenCombine/Publishers/Publishers.Output.swift b/Sources/OpenCombine/Publishers/Publishers.Output.swift
index ba1ba94..e2debfa 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Output.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Output.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 24.10.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Republishes elements up to the specified maximum count.
diff --git a/Sources/OpenCombine/Publishers/Publishers.Print.swift b/Sources/OpenCombine/Publishers/Publishers.Print.swift
index 9347948..77ac222 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Print.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Print.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 16.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Prints log messages for all publishing events.
diff --git a/Sources/OpenCombine/Publishers/Publishers.ReceiveOn.swift b/Sources/OpenCombine/Publishers/Publishers.ReceiveOn.swift
index 8cdca56..87f1d17 100644
--- a/Sources/OpenCombine/Publishers/Publishers.ReceiveOn.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.ReceiveOn.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 02.12.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
     /// Specifies the scheduler on which to receive elements from the publisher.
     ///
@@ -101,7 +97,7 @@ extension Publishers.ReceiveOn {
 
         private let lock = UnfairLock.allocate()
         private var state: State
-        private let downstreamLock = UnfairLock.allocate()
+        private let downstreamLock = UnfairRecursiveLock.allocate()
 
         init(_ receiveOn: ReceiveOn, downstream: Downstream) {
             state = .ready(receiveOn, downstream)
@@ -121,18 +117,6 @@ extension Publishers.ReceiveOn {
             }
             state = .subscribed(receiveOn, downstream, subscription)
             lock.unlock()
-            receiveOn.scheduler.schedule(options: receiveOn.options) { [weak self] in
-                self?.scheduledReceive(subscription: subscription)
-            }
-        }
-
-        private func scheduledReceive(subscription: Subscription) {
-            lock.lock()
-            guard case let .subscribed(_, downstream, _) = state else {
-                lock.unlock()
-                return
-            }
-            lock.unlock()
             downstreamLock.lock()
             downstream.receive(subscription: self)
             downstreamLock.unlock()
diff --git a/Sources/OpenCombine/Publishers/Publishers.ReplaceError.swift b/Sources/OpenCombine/Publishers/Publishers.ReplaceError.swift
index 3de0b75..6dcead8 100644
--- a/Sources/OpenCombine/Publishers/Publishers.ReplaceError.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.ReplaceError.swift
@@ -5,10 +5,6 @@
 //  Created by Bogdan Vlad on 8/29/19.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
     /// Replaces any errors in the stream with the provided element.
     ///
diff --git a/Sources/OpenCombine/Publishers/Publishers.Scan.swift b/Sources/OpenCombine/Publishers/Publishers.Scan.swift
index 3a96a8c..4d24b15 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Scan.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Scan.swift
@@ -4,10 +4,6 @@
 //  Created by Eric Patey on 26.08.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Transforms elements from the upstream publisher by providing the current element
diff --git a/Sources/OpenCombine/Publishers/Publishers.Sequence.swift b/Sources/OpenCombine/Publishers/Publishers.Sequence.swift
index 0036d4f..0a21198 100644
--- a/Sources/OpenCombine/Publishers/Publishers.Sequence.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.Sequence.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 19.06.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publishers {
 
     /// A publisher that publishes a given sequence of elements.
diff --git a/Sources/OpenCombine/Publishers/Publishers.SubscribeOn.swift b/Sources/OpenCombine/Publishers/Publishers.SubscribeOn.swift
index 7b4f02f..5bbfeea 100644
--- a/Sources/OpenCombine/Publishers/Publishers.SubscribeOn.swift
+++ b/Sources/OpenCombine/Publishers/Publishers.SubscribeOn.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 02.12.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 extension Publisher {
 
     /// Specifies the scheduler on which to perform subscribe, cancel, and request
diff --git a/Sources/OpenCombine/Publishers/Record.swift b/Sources/OpenCombine/Publishers/Record.swift
index daa824f..1ad5181 100644
--- a/Sources/OpenCombine/Publishers/Record.swift
+++ b/Sources/OpenCombine/Publishers/Record.swift
@@ -5,10 +5,6 @@
 //  Created by Sergej Jaskiewicz on 12.11.2019.
 //
 
-#if canImport(COpenCombineHelpers)
-import COpenCombineHelpers
-#endif
-
 /// A publisher that allows for recording a series of inputs and a completion for later
 /// playback to each subscriber.
 public struct Record<Output, Failure: Error>: Publisher {
diff --git a/Sources/OpenCombine/Subscribers/Subscribers.Demand.swift b/Sources/OpenCombine/Subscribers/Subscribers.Demand.swift
index 9451278..90e1f38 100644
--- a/Sources/OpenCombine/Subscribers/Subscribers.Demand.swift
+++ b/Sources/OpenCombine/Subscribers/Subscribers.Demand.swift
@@ -330,6 +330,7 @@ extension Subscribers {
 
         /// Returns `true` if `lhs` and `rhs` are not equal. `.unlimited` is not equal to
         /// any integer.
+        @inlinable
         public static func != (lhs: Demand, rhs: Int) -> Bool {
             if lhs == .unlimited {
                 return true
@@ -340,6 +341,7 @@ extension Subscribers {
 
         /// Returns `true` if `lhs` and `rhs` are equal. `.unlimited` is not equal to any
         /// integer.
+        @inlinable
         public static func == (lhs: Int, rhs: Demand) -> Bool {
             if rhs == .unlimited {
                 return false
@@ -350,6 +352,7 @@ extension Subscribers {
 
         /// Returns `true` if `lhs` and `rhs` are not equal. `.unlimited` is not equal to
         /// any integer.
+        @inlinable
         public static func != (lhs: Int, rhs: Demand) -> Bool {
             if rhs == .unlimited {
                 return true
@@ -358,8 +361,13 @@ extension Subscribers {
             }
         }
 
+        @inlinable
+        public static func == (lhs: Demand, rhs: Demand) -> Bool {
+            return lhs.rawValue == rhs.rawValue
+        }
+
         /// Returns the number of requested values, or `nil` if `.unlimited`.
-        public var max: Int? {
+        @inlinable public var max: Int? {
             if self == .unlimited {
                 return nil
             } else {
diff --git a/Sources/OpenCombineDispatch/DispatchQueue+Scheduler.swift b/Sources/OpenCombineDispatch/DispatchQueue+Scheduler.swift
index 90dfc08..a3e6724 100644
--- a/Sources/OpenCombineDispatch/DispatchQueue+Scheduler.swift
+++ b/Sources/OpenCombineDispatch/DispatchQueue+Scheduler.swift
@@ -126,12 +126,42 @@ extension DispatchQueue {
                     case .nanoseconds(let nanoseconds):
                         self = .nanoseconds(nanoseconds)
                     case .never:
-                        fallthrough
-                    @unknown default:
                         self = .nanoseconds(.max)
+                    @unknown default:
+                        self.init(__guessFromUnknown: timeInterval)
                     }
                 }
 
+                public // testable
+                init(__guessFromUnknown timeInterval: DispatchTimeInterval) {
+                    // Let's take some reference time,
+                    // add `timeInterval` to it, take the `rawValue` from the result
+                    // and subtract the `rawValue` of the reference time.
+                    //
+                    // We won't be able to provide the exact implementation though,
+                    // because something will definitely overflow.
+                    //
+                    // However, we can try to support as wide a range of values
+                    // as possible.
+                    //
+                    // By trial and error I got that the `rawValue` of `UInt64.max / 13`
+                    // gives us propably the widest range of supported values:
+                    // from `Int.min / 7` to `Int.max / 3` nanoseconds.
+                    // That's with Int being 64 bits. Since here only UInt64 can overflow,
+                    // when Int is 32 bits, we don't have this issue.
+                    // It should be more than enough.
+
+                    let referenceTime = DispatchTime(uptimeNanoseconds: .max / 13)
+
+                    let (partial, overflow) = (referenceTime + timeInterval).rawValue
+                        .subtractingReportingOverflow(referenceTime.rawValue)
+
+                    let result = overflow
+                        ? -Int(.max &- partial + 1) // Dont' ask.
+                        : Int(partial)
+                    self = .nanoseconds(result)
+                }
+
                 /// Creates a dispatch queue time interval from a floating-point
                 /// seconds value.
                 ///
@@ -191,15 +221,15 @@ extension DispatchQueue {
                 }
 
                 public static func seconds(_ value: Int) -> Stride {
-                    return Stride(magnitude: value * 1_000_000_000)
+                    return Stride(magnitude: clampedIntProduct(value, 1_000_000_000))
                 }
 
                 public static func milliseconds(_ value: Int) -> Stride {
-                    return Stride(magnitude: value * 1_000_000)
+                    return Stride(magnitude: clampedIntProduct(value, 1_000_000))
                 }
 
                 public static func microseconds(_ value: Int) -> Stride {
-                    return Stride(magnitude: value * 1_000)
+                    return Stride(magnitude: clampedIntProduct(value, 1_000))
                 }
 
                 public static func nanoseconds(_ value: Int) -> Stride {
@@ -336,3 +366,18 @@ extension DispatchQueue: OpenCombine.Scheduler {
     }
 }
 #endif
+
+// This function is taken from swift-corlibs-libdispatch:
+// https://github.com/apple/swift-corelibs-libdispatch/blob/c992dacf3ca114806e6ac9ffc9113b19255be9fe/src/swift/Time.swift#L134-L144
+//
+// Returns m1 * m2, clamped to the range [Int.min, Int.max].
+// Because of the way this function is used, we can always assume
+// that m2 > 0.
+private func clampedIntProduct(_ lhs: Int, _ rhs: Int) -> Int {
+    assert(rhs > 0, "multiplier must be positive")
+    let (result, overflow) = lhs.multipliedReportingOverflow(by: rhs)
+    if overflow {
+        return lhs > 0 ? .max : .min
+    }
+    return result
+}
diff --git a/Sources/OpenCombineFoundation/Helpers/Locking.swift b/Sources/OpenCombineFoundation/Helpers/Locking.swift
new file mode 100644
index 0000000..e7cd445
--- /dev/null
+++ b/Sources/OpenCombineFoundation/Helpers/Locking.swift
@@ -0,0 +1,15 @@
+//
+//  Locking.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+#if canImport(COpenCombineHelpers)
+import COpenCombineHelpers
+#endif
+
+import OpenCombine
+
+internal typealias UnfairLock = __UnfairLock
+internal typealias UnfairRecursiveLock = __UnfairRecursiveLock
diff --git a/Sources/OpenCombineFoundation/Helpers/Violations.swift b/Sources/OpenCombineFoundation/Helpers/Violations.swift
new file mode 100644
index 0000000..87ce60d
--- /dev/null
+++ b/Sources/OpenCombineFoundation/Helpers/Violations.swift
@@ -0,0 +1,17 @@
+//
+//  Violations.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 13.12.2019.
+//
+
+import OpenCombine
+
+extension Subscribers.Demand {
+    internal func assertNonZero(file: StaticString = #file,
+                                line: UInt = #line) {
+        if self == .none {
+            fatalError("API Violation: demand must not be zero", file: file, line: line)
+        }
+    }
+}
diff --git a/Sources/OpenCombineFoundation/JSONEncoder.swift b/Sources/OpenCombineFoundation/JSONEncoder.swift
new file mode 100644
index 0000000..5cb3ff6
--- /dev/null
+++ b/Sources/OpenCombineFoundation/JSONEncoder.swift
@@ -0,0 +1,17 @@
+//
+//  JSONEncoder.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.10.2019.
+//
+
+import Foundation
+import OpenCombine
+
+extension JSONEncoder: TopLevelEncoder {
+  public typealias Output = Data
+}
+
+extension JSONDecoder: TopLevelDecoder {
+  public typealias Input = Data
+}
diff --git a/Sources/OpenCombineFoundation/NotificationCenter.swift b/Sources/OpenCombineFoundation/NotificationCenter.swift
new file mode 100644
index 0000000..f108fc3
--- /dev/null
+++ b/Sources/OpenCombineFoundation/NotificationCenter.swift
@@ -0,0 +1,230 @@
+//
+//  NotificationCenter.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.10.2019.
+//
+
+import Foundation
+import OpenCombine
+
+extension NotificationCenter {
+
+    /// A namespace for disambiguation when both OpenCombine and Foundation are imported.
+    ///
+    /// Foundation extends `NotificationCenter` with new methods and nested types.
+    /// If you import both OpenCombine and Foundation, you will not be able
+    /// to write `NotificationCenter.Publisher`,
+    /// because Swift is unable to understand which `Publisher`
+    /// you're referring to — the one declared in Foundation or in OpenCombine.
+    ///
+    /// So you have to write `NotificationCenter.OCombine.Publisher`.
+    ///
+    /// This bug is tracked [here](https://bugs.swift.org/browse/SR-11183).
+    ///
+    /// You can omit this whenever Combine is not available (e. g. on Linux).
+    public struct OCombine {
+
+        public let center: NotificationCenter
+
+        public init(_ center: NotificationCenter) {
+            self.center = center
+        }
+
+        /// A publisher that emits elements when broadcasting notifications.
+        public struct Publisher: OpenCombine.Publisher {
+
+            public typealias Output = Notification
+
+            public typealias Failure = Never
+
+            /// The notification center this publisher uses as a source.
+            public let center: NotificationCenter
+
+            /// The name of notifications published by this publisher.
+            public let name: Notification.Name
+
+            /// The object posting the named notfication.
+            public let object: AnyObject?
+
+            /// Creates a publisher that emits events when broadcasting notifications.
+            ///
+            /// - Parameters:
+            ///   - center: The notification center to publish notifications for.
+            ///   - name: The name of the notification to publish.
+            ///   - object: The object posting the named notfication. If `nil`,
+            ///     the publisher emits elements for any object producing a notification
+            ///     with the given name.
+            public init(center: NotificationCenter,
+                        name: Notification.Name,
+                        object: AnyObject? = nil) {
+                self.center = center
+                self.name = name
+                self.object = object
+            }
+
+            public func receive<Downstream: Subscriber>(subscriber: Downstream)
+                where Downstream.Failure == Never, Downstream.Input == Notification
+            {
+                let subscription = Notification.Subscription(center: center,
+                                                             name: name,
+                                                             object: object,
+                                                             downstream: subscriber)
+                subscriber.receive(subscription: subscription)
+            }
+        }
+
+        /// Returns a publisher that emits events when broadcasting notifications.
+        ///
+        /// - Parameters:
+        ///   - name: The name of the notification to publish.
+        ///   - object: The object posting the named notfication. If `nil`, the publisher
+        ///     emits elements for any object producing a notification with the given
+        ///     name.
+        /// - Returns: A publisher that emits events when broadcasting notifications.
+        public func publisher(for name: Notification.Name,
+                              object: AnyObject? = nil) -> Publisher {
+            return .init(center: center, name: name, object: object)
+        }
+    }
+
+#if !canImport(Combine)
+    /// A publisher that emits elements when broadcasting notifications.
+    public typealias Publisher = OCombine.Publisher
+#endif
+}
+
+extension NotificationCenter {
+
+    /// A namespace for disambiguation when both OpenCombine and Foundation are imported.
+    ///
+    /// Foundation extends `NotificationCenter` with new methods and nested types.
+    /// If you import both OpenCombine and Foundation, you will not be able
+    /// to write `NotificationCenter.default.publisher(for: name)`,
+    /// because Swift is unable to understand which `publisher` method
+    /// you're referring to — the one declared in Foundation or in OpenCombine.
+    ///
+    /// So you have to write `NotificationCenter.default.ocombine.publisher(for: name)`.
+    ///
+    /// This bug is tracked [here](https://bugs.swift.org/browse/SR-11183).
+    ///
+    /// You can omit this whenever Combine is not available (e. g. on Linux).
+    public var ocombine: OCombine { return .init(self) }
+
+#if !canImport(Combine)
+    /// Returns a publisher that emits events when broadcasting notifications.
+    ///
+    /// - Parameters:
+    ///   - name: The name of the notification to publish.
+    ///   - object: The object posting the named notfication. If `nil`, the publisher
+    ///     emits elements for any object producing a notification with the given name.
+    /// - Returns: A publisher that emits events when broadcasting notifications.
+    public func publisher(for name: Notification.Name,
+                          object: AnyObject? = nil) -> OCombine.Publisher {
+        return ocombine.publisher(for: name, object: object)
+    }
+#endif
+}
+
+extension NotificationCenter.OCombine.Publisher: Equatable {
+    public static func == (lhs: NotificationCenter.OCombine.Publisher,
+                           rhs: NotificationCenter.OCombine.Publisher) -> Bool {
+        return lhs.center == rhs.center &&
+               lhs.name == rhs.name &&
+               lhs.object === rhs.object
+    }
+}
+
+extension Notification {
+    fileprivate final class Subscription<Downstream: Subscriber>
+        : OpenCombine.Subscription,
+          CustomStringConvertible,
+          CustomReflectable,
+          CustomPlaygroundDisplayConvertible
+        where Downstream.Input == Notification, Downstream.Failure == Never
+    {
+        private let lock = UnfairLock.allocate()
+
+        fileprivate let downstreamLock = UnfairRecursiveLock.allocate()
+
+        fileprivate var demand = Subscribers.Demand.none
+
+        private var center: NotificationCenter?
+
+        private let name: Name
+
+        private var object: AnyObject?
+
+        private var observation: AnyObject?
+
+        fileprivate init(center: NotificationCenter,
+                         name: Notification.Name,
+                         object: AnyObject?,
+                         downstream: Downstream) {
+            self.center = center
+            self.name = name
+            self.object = object
+            self.observation = center
+                .addObserver(forName: name, object: object, queue: nil) { [weak self] in
+                    self?.didReceiveNotification($0, downstream: downstream)
+                }
+        }
+
+        deinit {
+            lock.deallocate()
+            downstreamLock.deallocate()
+        }
+
+        private func didReceiveNotification(_ notification: Notification,
+                                            downstream: Downstream) {
+            lock.lock()
+            guard demand > 0 else {
+                lock.unlock()
+                return
+            }
+            demand -= 1
+            lock.unlock()
+            downstreamLock.lock()
+            let newDemand = downstream.receive(notification)
+            downstreamLock.unlock()
+            lock.lock()
+            demand += newDemand
+            lock.unlock()
+        }
+
+        func request(_ demand: Subscribers.Demand) {
+            lock.lock()
+            self.demand += demand
+            lock.unlock()
+        }
+
+        func cancel() {
+            lock.lock()
+            guard let center = self.center, let observation = self.observation else {
+                lock.unlock()
+                return
+            }
+            self.center = nil
+            self.object = nil
+            self.observation = nil
+            lock.unlock()
+            center.removeObserver(observation)
+        }
+
+        fileprivate var description: String { return "NotificationCenter Observer" }
+
+        fileprivate var customMirror: Mirror {
+            lock.lock()
+            defer { lock.unlock() }
+            let children: [Mirror.Child] = [
+                ("center", center as Any),
+                ("name", name),
+                ("object", object as Any),
+                ("demand", demand)
+            ]
+            return Mirror(self, children: children)
+        }
+
+        fileprivate var playgroundDescription: Any { return description }
+    }
+}
diff --git a/Sources/OpenCombineFoundation/PropertyListEncoder.swift b/Sources/OpenCombineFoundation/PropertyListEncoder.swift
new file mode 100644
index 0000000..4694c10
--- /dev/null
+++ b/Sources/OpenCombineFoundation/PropertyListEncoder.swift
@@ -0,0 +1,17 @@
+//
+//  PropertyListEncoder.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import OpenCombine
+
+extension PropertyListEncoder: TopLevelEncoder {
+  public typealias Output = Data
+}
+
+extension PropertyListDecoder: TopLevelDecoder {
+  public typealias Input = Data
+}
diff --git a/Sources/OpenCombineFoundation/URLSession.swift b/Sources/OpenCombineFoundation/URLSession.swift
new file mode 100644
index 0000000..8bcab45
--- /dev/null
+++ b/Sources/OpenCombineFoundation/URLSession.swift
@@ -0,0 +1,233 @@
+//
+//  URLSession.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 13.12.2019.
+//
+
+import Foundation
+
+#if canImport(FoundationNetworking)
+import FoundationNetworking
+#endif
+
+import OpenCombine
+
+extension URLSession {
+
+    /// A namespace for disambiguation when both OpenCombine and Foundation are imported.
+    ///
+    /// Foundation extends `URLSession` with new methods and nested types.
+    /// If you import both OpenCombine and Foundation, you will not be able
+    /// to write `URLSession.DataTaskPublisher`,
+    /// because Swift is unable to understand which `DataTaskPublisher`
+    /// you're referring to — the one declared in Foundation or in OpenCombine.
+    ///
+    /// So you have to write `URLSession.OCombine.DataTaskPublisher`.
+    ///
+    /// This bug is tracked [here](https://bugs.swift.org/browse/SR-11183).
+    ///
+    /// You can omit this whenever Combine is not available (e. g. on Linux).
+    public struct OCombine {
+
+        public let session: URLSession
+
+        public init(_ session: URLSession) {
+            self.session = session
+        }
+
+        public struct DataTaskPublisher: Publisher {
+
+            public typealias Output = (data: Data, response: URLResponse)
+
+            public typealias Failure = URLError
+
+            public let request: URLRequest
+
+            public let session: URLSession
+
+            public init(request: URLRequest, session: URLSession) {
+                self.request = request
+                self.session = session
+            }
+
+            public func receive<Downstream: Subscriber>(subscriber: Downstream)
+                where Downstream.Failure == Failure, Downstream.Input == Output
+            {
+                let subscription = Inner(parent: self, downstream: subscriber)
+                subscriber.receive(subscription: subscription)
+            }
+        }
+
+        /// Returns a publisher that wraps a URL session data task for a given URL.
+        ///
+        /// The publisher publishes data when the task completes, or terminates if
+        /// the task fails with an error.
+        ///
+        /// - Parameter url: The URL for which to create a data task.
+        /// - Returns: A publisher that wraps a data task for the URL.
+        public func dataTaskPublisher(for url: URL) -> DataTaskPublisher {
+            return dataTaskPublisher(for: URLRequest(url: url))
+        }
+
+        /// Returns a publisher that wraps a URL session data task for a given
+        /// URL request.
+        ///
+        /// The publisher publishes data when the task completes, or terminates if
+        /// the task fails with an error.
+        ///
+        /// - Parameter request: The URL request for which to create a data task.
+        /// - Returns: A publisher that wraps a data task for the URL request.
+        public func dataTaskPublisher(for request: URLRequest) -> DataTaskPublisher {
+            return .init(request: request, session: session)
+        }
+    }
+
+#if !canImport(Combine)
+    public typealias DataTaskPublisher = OCombine.DataTaskPublisher
+#endif
+}
+
+extension URLSession {
+
+    /// A namespace for disambiguation when both OpenCombine and Foundation are imported.
+    ///
+    /// Foundation extends `URLSession` with new methods and nested types.
+    /// If you import both OpenCombine and Foundation, you will not be able
+    /// to write `URLSession.shared.dataTaskPublisher(for: url)`,
+    /// because Swift is unable to understand which `dataTaskPublisher` method
+    /// you're referring to — the one declared in Foundation or in OpenCombine.
+    ///
+    /// So you have to write `URLSession.shared.ocombine.dataTaskPublisher(for: url)`.
+    ///
+    /// This bug is tracked [here](https://bugs.swift.org/browse/SR-11183).
+    ///
+    /// You can omit this whenever Combine is not available (e. g. on Linux).
+    public var ocombine: OCombine { return .init(self) }
+
+#if !canImport(Combine)
+    /// Returns a publisher that wraps a URL session data task for a given URL.
+    ///
+    /// The publisher publishes data when the task completes, or terminates if the task
+    /// fails with an error.
+    ///
+    /// - Parameter url: The URL for which to create a data task.
+    /// - Returns: A publisher that wraps a data task for the URL.
+    public func dataTaskPublisher(for url: URL) -> DataTaskPublisher {
+        return ocombine.dataTaskPublisher(for: url)
+    }
+
+    /// Returns a publisher that wraps a URL session data task for a given URL request.
+    ///
+    /// The publisher publishes data when the task completes, or terminates if the task
+    /// fails with an error.
+    ///
+    /// - Parameter request: The URL request for which to create a data task.
+    /// - Returns: A publisher that wraps a data task for the URL request.
+    public func dataTaskPublisher(for request: URLRequest) -> DataTaskPublisher {
+        return ocombine.dataTaskPublisher(for: request)
+    }
+#endif
+}
+
+extension URLSession.OCombine.DataTaskPublisher {
+    private class Inner<Downstream: Subscriber>
+        : Subscription,
+          CustomStringConvertible,
+          CustomReflectable,
+          CustomPlaygroundDisplayConvertible
+        where Downstream.Input == (data: Data, response: URLResponse),
+              Downstream.Failure == URLError
+    {
+        private let lock = UnfairLock.allocate()
+
+        private var parent: URLSession.OCombine.DataTaskPublisher?
+
+        private var downstream: Downstream?
+
+        private var demand = Subscribers.Demand.none
+
+        private var task: URLSessionDataTask?
+
+        fileprivate init(parent: URLSession.OCombine.DataTaskPublisher,
+                         downstream: Downstream) {
+            self.parent = parent
+            self.downstream = downstream
+        }
+
+        deinit {
+            lock.deallocate()
+        }
+
+        func request(_ demand: Subscribers.Demand) {
+            demand.assertNonZero()
+            lock.lock()
+            guard let parent = self.parent else {
+                lock.unlock()
+                return
+            }
+            if self.task == nil {
+                task = parent.session.dataTask(with: parent.request,
+                                               completionHandler: handleResponse)
+            }
+            self.demand += demand
+            let task = self.task
+            lock.unlock()
+            task?.resume()
+        }
+
+        private func handleResponse(data: Data?, response: URLResponse?, error: Error?) {
+            lock.lock()
+            guard demand > 0, parent != nil, let downstream = self.downstream else {
+                lock.unlock()
+                return
+            }
+            lockedTerminate()
+            lock.unlock()
+            switch (data, response, error) {
+            case let (data, response?, nil):
+                _ = downstream.receive((data ?? Data(), response))
+                downstream.receive(completion: .finished)
+            case let (_, _, error as URLError):
+                downstream.receive(completion: .failure(error))
+            default:
+                downstream.receive(completion: .failure(URLError(.unknown)))
+            }
+        }
+
+        func cancel() {
+            lock.lock()
+            guard parent != nil else {
+                lock.unlock()
+                return
+            }
+            let task = self.task
+            lockedTerminate()
+            lock.unlock()
+            task?.cancel()
+        }
+
+        private func lockedTerminate() {
+            parent = nil
+            downstream = nil
+            demand = .none
+            task = nil
+        }
+
+        var description: String { return "DataTaskPublisher" }
+
+        var customMirror: Mirror {
+            lock.lock()
+            defer { lock.unlock() }
+            let children: [Mirror.Child] = [
+                ("task", task as Any),
+                ("downstream", downstream as Any),
+                ("parent", parent as Any),
+                ("demand", demand)
+            ]
+            return Mirror(self, children: children)
+        }
+
+        var playgroundDescription: Any { return description }
+    }
+}


Test patch:
diff --git a/Tests/OpenCombineTests/DispatchTests/DispatchQueueSchedulerTests.swift b/Tests/OpenCombineTests/DispatchTests/DispatchQueueSchedulerTests.swift
index e422c6e..ae35661 100644
--- a/Tests/OpenCombineTests/DispatchTests/DispatchQueueSchedulerTests.swift
+++ b/Tests/OpenCombineTests/DispatchTests/DispatchQueueSchedulerTests.swift
@@ -84,8 +84,6 @@ final class DispatchQueueSchedulerTests: XCTestCase {
     // MARK: - Scheduler.SchedulerTimeType.Stride
 
     func testStrideToDispatchTimeInterval() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
         switch (Stride.seconds(2).timeInterval,
                 Stride.milliseconds(2).timeInterval,
                 Stride.microseconds(2).timeInterval,
@@ -100,25 +98,117 @@ final class DispatchQueueSchedulerTests: XCTestCase {
         }
     }
 
-    func testStrideFromDispatchTimeInterval() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
+    func testStrideFromDispatchTimeInterval() throws {
         XCTAssertEqual(Stride(.seconds(2)).magnitude, 2_000_000_000)
         XCTAssertEqual(Stride(.milliseconds(2)).magnitude, 2_000_000)
         XCTAssertEqual(Stride(.microseconds(2)).magnitude, 2_000)
         XCTAssertEqual(Stride(.nanoseconds(2)).magnitude, 2)
+
         XCTAssertEqual(Stride(.never).magnitude, .max)
+        XCTAssertEqual(Stride(.nanoseconds(.max)).magnitude, .max)
+        XCTAssertEqual(Stride(.nanoseconds(.min)).magnitude, .min)
+        XCTAssertEqual(Stride(.microseconds(.max)).magnitude, .max)
+        XCTAssertEqual(Stride(.microseconds(.min)).magnitude, .min)
+        XCTAssertEqual(Stride(.milliseconds(.max)).magnitude, .max)
+        XCTAssertEqual(Stride(.milliseconds(.min)).magnitude, .min)
+        XCTAssertEqual(Stride(.seconds(.max)).magnitude, .max)
+        XCTAssertEqual(Stride(.seconds(.min)).magnitude, .min)
     }
 
-    func testStrideFromNumericValue() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
+    func testStrideFromUnknownDispatchTimeIntervalCase() {
+        // Here we're testing out internal API that is not present in Combine.
+        // Although we prefer only testing public APIs, this case is special.
+        let makeStride: (DispatchTimeInterval) -> Stride
+#if OPENCOMBINE_COMPATIBILITY_TEST
+        makeStride = Stride.init(_:)
+#else
+        makeStride = Stride.init(__guessFromUnknown:)
+#endif
+        XCTAssertEqual(makeStride(.nanoseconds(.min / 7)).magnitude, .min / 7)
+        XCTAssertEqual(makeStride(.nanoseconds(-128)).magnitude, -128)
+        XCTAssertEqual(makeStride(.nanoseconds(-57)).magnitude, -57)
+        XCTAssertEqual(makeStride(.nanoseconds(-33)).magnitude, -33)
+        XCTAssertEqual(makeStride(.nanoseconds(-17)).magnitude, -17)
+        XCTAssertEqual(makeStride(.nanoseconds(-8)).magnitude, -8)
+        XCTAssertEqual(makeStride(.nanoseconds(-3)).magnitude, -3)
+        XCTAssertEqual(makeStride(.nanoseconds(-1)).magnitude, -1)
+        XCTAssertEqual(makeStride(.nanoseconds(0)).magnitude, 0)
+        XCTAssertEqual(makeStride(.nanoseconds(1)).magnitude, 1)
+        XCTAssertEqual(makeStride(.nanoseconds(3)).magnitude, 3)
+        XCTAssertEqual(makeStride(.nanoseconds(8)).magnitude, 8)
+        XCTAssertEqual(makeStride(.nanoseconds(17)).magnitude, 17)
+        XCTAssertEqual(makeStride(.nanoseconds(33)).magnitude, 33)
+        XCTAssertEqual(makeStride(.nanoseconds(57)).magnitude, 57)
+        XCTAssertEqual(makeStride(.nanoseconds(128)).magnitude, 128)
+        XCTAssertEqual(makeStride(.nanoseconds(.max / 3)).magnitude, .max / 3)
+
+        XCTAssertEqual(makeStride(.microseconds(-128)).magnitude, -128_000)
+        XCTAssertEqual(makeStride(.microseconds(-57)).magnitude, -57_000)
+        XCTAssertEqual(makeStride(.microseconds(-33)).magnitude, -33_000)
+        XCTAssertEqual(makeStride(.microseconds(-17)).magnitude, -17_000)
+        XCTAssertEqual(makeStride(.microseconds(-8)).magnitude, -8_000)
+        XCTAssertEqual(makeStride(.microseconds(-3)).magnitude, -3_000)
+        XCTAssertEqual(makeStride(.microseconds(-1)).magnitude, -1_000)
+        XCTAssertEqual(makeStride(.microseconds(0)).magnitude, 0)
+        XCTAssertEqual(makeStride(.microseconds(1)).magnitude, 1_000)
+        XCTAssertEqual(makeStride(.microseconds(3)).magnitude, 3_000)
+        XCTAssertEqual(makeStride(.microseconds(8)).magnitude, 8_000)
+        XCTAssertEqual(makeStride(.microseconds(17)).magnitude, 17_000)
+        XCTAssertEqual(makeStride(.microseconds(33)).magnitude, 33_000)
+        XCTAssertEqual(makeStride(.microseconds(57)).magnitude, 57_000)
+        XCTAssertEqual(makeStride(.microseconds(128)).magnitude, 128_000)
+
+        XCTAssertEqual(makeStride(.milliseconds(-128)).magnitude, -128_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-57)).magnitude, -57_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-33)).magnitude, -33_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-17)).magnitude, -17_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-8)).magnitude, -8_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-3)).magnitude, -3_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(-1)).magnitude, -1_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(0)).magnitude, 0)
+        XCTAssertEqual(makeStride(.milliseconds(1)).magnitude, 1_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(3)).magnitude, 3_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(8)).magnitude, 8_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(17)).magnitude, 17_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(33)).magnitude, 33_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(57)).magnitude, 57_000_000)
+        XCTAssertEqual(makeStride(.milliseconds(128)).magnitude, 128_000_000)
+
+        XCTAssertEqual(makeStride(.seconds(-2)).magnitude, -2_000_000_000)
+        XCTAssertEqual(makeStride(.seconds(-1)).magnitude, -1_000_000_000)
+        XCTAssertEqual(makeStride(.seconds(0)).magnitude, 0)
+        XCTAssertEqual(makeStride(.seconds(1)).magnitude, 1_000_000_000)
+        XCTAssertEqual(makeStride(.seconds(2)).magnitude, 2_000_000_000)
+    }
 
+    func testStrideFromNumericValue() {
         XCTAssertEqual(Stride.seconds(1.2).magnitude, 1_200_000_000)
         XCTAssertEqual(Stride.seconds(2).magnitude, 2_000_000_000)
         XCTAssertEqual(Stride.milliseconds(2).magnitude, 2_000_000)
         XCTAssertEqual(Stride.microseconds(2).magnitude, 2_000)
         XCTAssertEqual(Stride.nanoseconds(2).magnitude, 2)
 
+#if arch(x86_64) || arch(arm64) || arch(s390x) || arch(powerpc64) || arch(powerpc64le)
+        // 64-bit platforms
+        XCTAssertEqual(
+            Stride.seconds(Double(Int.max) / 1_000_000_000 - 1).magnitude,
+            9223372035854776320
+        )
+#elseif arch(i386) || arch(arm)
+        // 32-bit platforms
+        XCTAssertEqual(
+            Stride.seconds(Double(Int.max) / 1_000_000_000).magnitude,
+            .max
+        )
+#else
+#error("This architecture isn't known. Add it to the 32-bit or 64-bit line.")
+#endif
+
+        XCTAssertEqual(Stride.seconds(.max).magnitude, .max)
+        XCTAssertEqual(Stride.milliseconds(.max).magnitude, .max)
+        XCTAssertEqual(Stride.microseconds(.max).magnitude, .max)
+        XCTAssertEqual(Stride.nanoseconds(.max).magnitude, .max)
+
         XCTAssertEqual((1.2 as Stride).magnitude, 1_200_000_000)
         XCTAssertEqual((2 as Stride).magnitude, 2_000_000_000)
 
@@ -126,17 +216,33 @@ final class DispatchQueueSchedulerTests: XCTestCase {
         XCTAssertEqual(Stride(exactly: 871 as UInt64)?.magnitude, 871)
     }
 
-    func testStrideComparable() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
+    func testStrideFromTooMuchSecondsCrashes() {
+        assertCrashes {
+#if arch(x86_64) || arch(arm64) || arch(s390x) || arch(powerpc64) || arch(powerpc64le)
+            // 64-bit platforms
+            XCTAssertGreaterThan(
+                Stride.seconds(Double(Int.max) / 1_000_000_000).magnitude,
+                .max
+            )
+#elseif arch(i386) || arch(arm)
+            // 32-bit platforms
+            XCTAssertGreaterThan(
+                Stride.seconds(Double(Int.max) / 1_000_000_000 + 1).magnitude,
+                .max
+            )
+#else
+#error("This architecture isn't known. Add it to the 32-bit or 64-bit line.")
+#endif
+        }
+    }
 
+    func testStrideComparable() {
         XCTAssertLessThan(Stride.nanoseconds(1), .nanoseconds(2))
         XCTAssertGreaterThan(Stride.nanoseconds(-2), .microseconds(-10))
         XCTAssertLessThan(Stride.milliseconds(2), .seconds(2))
     }
 
     func testStrideMultiplication() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
         XCTAssertEqual((Stride.nanoseconds(0) * .nanoseconds(61346)).magnitude, 0)
         XCTAssertEqual((Stride.nanoseconds(61346) * .nanoseconds(0)).magnitude, 0)
         XCTAssertEqual((Stride.nanoseconds(18) * .nanoseconds(1)).magnitude, 18)
@@ -196,8 +302,6 @@ final class DispatchQueueSchedulerTests: XCTestCase {
     }
 
     func testStrideAddition() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
         XCTAssertEqual((Stride.nanoseconds(0) + .microseconds(2)).magnitude, 2000)
         XCTAssertEqual((Stride.nanoseconds(2) + .microseconds(0)).magnitude, 2)
         XCTAssertEqual((Stride.nanoseconds(7) + .nanoseconds(12)).magnitude, 19)
@@ -243,8 +347,6 @@ final class DispatchQueueSchedulerTests: XCTestCase {
     }
 
     func testStrideSubtraction() {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
         XCTAssertEqual((Stride.nanoseconds(0) - .microseconds(2)).magnitude, -2000)
         XCTAssertEqual((Stride.nanoseconds(2) - .microseconds(0)).magnitude, 2)
         XCTAssertEqual((Stride.nanoseconds(7) - .nanoseconds(12)).magnitude, -5)
@@ -290,8 +392,6 @@ final class DispatchQueueSchedulerTests: XCTestCase {
     }
 
     func testStrideCodable() throws {
-        typealias Stride = Scheduler.SchedulerTimeType.Stride
-
         let encoder = JSONEncoder()
         let decoder = JSONDecoder()
 
@@ -429,6 +529,9 @@ private let backgroundScheduler = DispatchQueue.global(qos: .background).ocombin
 
 #endif
 
+@available(macOS 10.15, iOS 13.0, *)
+private typealias Stride = Scheduler.SchedulerTimeType.Stride
+
 private struct KeyedWrapper<Value: Codable & Equatable>: Codable, Equatable {
     let value: Value
 }
diff --git a/Tests/OpenCombineTests/Extensions/TopLevelDecoder+Extensions.swift b/Tests/OpenCombineTests/Extensions/TopLevelDecoder+Extensions.swift
deleted file mode 100644
index cf1a3b6..0000000
--- a/Tests/OpenCombineTests/Extensions/TopLevelDecoder+Extensions.swift
+++ /dev/null
@@ -1,17 +0,0 @@
-//
-//  TopLevelDecoder+Extensions.swift
-//  
-//
-//  Created by Joseph Spadafora on 6/29/19.
-//
-
-#if !OPENCOMBINE_COMPATIBILITY_TEST
-import Foundation
-import OpenCombine
-
-extension JSONDecoder: TopLevelDecoder {}
-extension JSONEncoder: TopLevelEncoder {}
-
-extension PropertyListDecoder: TopLevelDecoder {}
-extension PropertyListEncoder: TopLevelEncoder {}
-#endif
diff --git a/Tests/OpenCombineTests/FoundationTests/JSONDecoderTests.swift b/Tests/OpenCombineTests/FoundationTests/JSONDecoderTests.swift
new file mode 100644
index 0000000..3cdaf48
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/JSONDecoderTests.swift
@@ -0,0 +1,63 @@
+//
+//  JSONDecoderTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class JSONDecoderTests: XCTestCase {
+    func testSuccessfullyDecode() {
+        let decoder = JSONDecoder()
+        let input = #"[{"success":true}]"#
+        var actualOutput: [Subscribers.Completion<TestingError>]?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .map { Data($0.utf8) }
+            .decode(type: [Subscribers.Completion<TestingError>].self, decoder: decoder)
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+        switch actualCompletion {
+        case .finished?:
+            XCTAssertEqual(actualOutput, [.finished])
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+
+    func testDecodingFailure() {
+        let decoder = JSONDecoder()
+        let input = #"{"a":1,"b":2}"#
+        var actualOutput: [Int]?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .map { Data($0.utf8) }
+            .decode(type: [Int].self, decoder: decoder)
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+        switch actualCompletion {
+        case .finished?:
+            XCTFail("Unexpected success")
+        case .failure(DecodingError.typeMismatch)?:
+            XCTAssertNil(actualOutput)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+}
diff --git a/Tests/OpenCombineTests/FoundationTests/JSONEncoderTests.swift b/Tests/OpenCombineTests/FoundationTests/JSONEncoderTests.swift
new file mode 100644
index 0000000..ecf5350
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/JSONEncoderTests.swift
@@ -0,0 +1,67 @@
+//
+//  JSONEncoderTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class JSONEncoderTests: XCTestCase {
+
+    func testSuccessfullyEncode() {
+        let encoder = JSONEncoder()
+        let input = [Subscribers.Completion<TestingError>.finished]
+        var actualOutput: String?
+        var actualCompletion: Subscribers.Completion<Error>?
+
+        let cancellable = Just(input)
+            .encode(encoder: encoder)
+            .map { String(decoding: $0, as: UTF8.self) }
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+
+        switch actualCompletion {
+        case .finished?:
+            XCTAssertEqual(actualOutput, #"[{"success":true}]"#)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+
+    func testEncodingFailure() {
+        let encoder = JSONEncoder()
+        let input = Double.nan
+        var actualOutput: String?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .encode(encoder: encoder)
+            .map { String(decoding: $0, as: UTF8.self) }
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+
+        switch actualCompletion {
+        case .finished?:
+            XCTFail("Unexpected success")
+        case .failure(EncodingError.invalidValue)?:
+            XCTAssertNil(actualOutput)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+}
diff --git a/Tests/OpenCombineTests/FoundationTests/NotificationCenterTests.swift b/Tests/OpenCombineTests/FoundationTests/NotificationCenterTests.swift
new file mode 100644
index 0000000..0066734
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/NotificationCenterTests.swift
@@ -0,0 +1,625 @@
+//
+//  NotificationCenterTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class NotificationCenterTests: XCTestCase {
+
+    func testRequestingDemand() {
+
+        let initialDemands: [Subscribers.Demand?] = [
+            nil,
+            .max(1),
+            .max(2),
+            .max(10),
+            .unlimited
+        ]
+
+        let subsequentDemands: [[Subscribers.Demand]] = [
+            Array(repeating: .max(0), count: 5),
+            Array(repeating: .max(1), count: 10),
+            [.max(1), .max(0), .max(1), .max(0)],
+            [.max(0), .max(1), .max(2)],
+            [.unlimited, .max(1)]
+        ]
+
+        var numberOfInputsHistory: [Int] = []
+        let expectedNumberOfInputsHistory = [
+            0, 0, 0, 0, 0, 1, 11, 2, 1, 20, 2, 12, 4, 5, 20, 10, 20, 12, 13, 20, 20,
+            20, 20, 20, 20
+        ]
+
+        for initialDemand in initialDemands {
+            for subsequentDemand in subsequentDemands {
+
+                var i = 0
+
+                let center = TestNotificationCenter()
+                let name = Notification.Name(rawValue: "testName")
+                let publisher = makePublisher(center, for: name, object: nil)
+
+                let subscriber = TrackingSubscriberBase<Notification, Never>(
+                    receiveSubscription: { initialDemand.map($0.request) },
+                    receiveValue: { _ in
+                        defer { i += 1 }
+                        return i < subsequentDemand.endIndex ? subsequentDemand[i] : .none
+                    }
+                )
+
+                XCTAssertEqual(subscriber.subscriptions.count, 0)
+                XCTAssertEqual(subscriber.inputs.count, 0)
+                XCTAssertEqual(subscriber.completions.count, 0)
+
+                publisher.subscribe(subscriber)
+
+                XCTAssertEqual(subscriber.subscriptions.count, 1)
+                XCTAssertEqual(subscriber.inputs.count, 0)
+                XCTAssertEqual(subscriber.completions.count, 0)
+
+                for _ in 0..<20 {
+                    center.post(name: name, object: TestObject.two)
+                }
+
+                XCTAssertEqual(subscriber.subscriptions.count, 1)
+                XCTAssertEqual(subscriber.completions.count, 0)
+
+                numberOfInputsHistory.append(subscriber.inputs.count)
+            }
+        }
+
+        XCTAssertEqual(numberOfInputsHistory, expectedNumberOfInputsHistory)
+    }
+
+    func testBasicBehavior() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let publisher = makePublisher(center, for: name, object: TestObject.one)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriberBase<Notification, Never>(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+
+        XCTAssertEqual(center.history, [])
+
+        publisher.subscribe(tracking)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, TestObject.one, nil)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer")])
+
+        let note = Notification(name: name, object: TestObject.one, userInfo: nil)
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, TestObject.one, nil),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer")])
+
+        try XCTUnwrap(downstreamSubscription).request(.max(3))
+
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note)])
+
+        let unrelatedNote1 = Notification(name: Notification.Name("unrelatedNote1"),
+                                          object: TestObject.one,
+                                          userInfo: nil)
+        center.post(unrelatedNote1)
+
+        let unrelatedNote2 = Notification(name: name,
+                                          object: TestObject.two,
+                                          userInfo: nil)
+        center.post(unrelatedNote2)
+        center.post(name: name, object: nil)
+        center.post(name: name, object: nil)
+        center.post(name: name, object: TestObject.one)
+        center.post(name: name, object: TestObject.one)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotification(note),
+                        .postNotification(unrelatedNote1),
+                        .postNotification(unrelatedNote2),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(Notification(name: name,
+                                                       object: TestObject.one)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(Notification(name: name,
+                                                       object: TestObject.one))])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note),
+                        .value(unrelatedNote1),
+                        .value(unrelatedNote2)])
+
+        try XCTUnwrap(downstreamSubscription).request(.unlimited)
+
+        center.post(note)
+
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotification(note),
+                        .postNotification(unrelatedNote1),
+                        .postNotification(unrelatedNote2),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(Notification(name: name,
+                                                       object: TestObject.one)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(Notification(name: name,
+                                                       object: TestObject.one)),
+                        .postNotification(note),
+                        .removeObserver,
+                        .removeObserverForName(nil, nil),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note),
+                        .value(unrelatedNote1),
+                        .value(unrelatedNote2),
+                        .value(note)])
+    }
+
+    func testBasicBehaviorNilObject() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let publisher = makePublisher(center, for: name, object: nil)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriberBase<Notification, Never>(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+
+        XCTAssertEqual(center.history, [])
+
+        publisher.subscribe(tracking)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer")])
+
+        let note = Notification(name: name, object: TestObject.one, userInfo: nil)
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer")])
+
+        try XCTUnwrap(downstreamSubscription).request(.max(3))
+
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .postNotification(note),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note)])
+
+        let unrelatedNote = Notification(name: Notification.Name("unrelatedNote"),
+                                         object: TestObject.one,
+                                         userInfo: nil)
+        center.post(unrelatedNote)
+        center.post(name: name, object: nil)
+        center.post(name: name, object: nil)
+        center.post(name: name, object: TestObject.one)
+        center.post(name: name, object: TestObject.one)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .postNotification(note),
+                        .postNotification(note),
+                        .postNotification(unrelatedNote),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note),
+                        .value(unrelatedNote),
+                        .value(Notification(name: name))])
+
+        try XCTUnwrap(downstreamSubscription).request(.unlimited)
+
+        center.post(note)
+
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        center.post(note)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .postNotification(note),
+                        .postNotification(note),
+                        .postNotification(unrelatedNote),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name)),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotificationWithName(name, TestObject.one, nil),
+                        .postNotification(note),
+                        .postNotification(note),
+                        .removeObserver,
+                        .removeObserverForName(nil, nil),
+                        .postNotification(note)])
+        XCTAssertEqual(tracking.history,
+                       [.subscription("NotificationCenter Observer"),
+                        .value(note),
+                        .value(unrelatedNote),
+                        .value(Notification(name: name)),
+                        .value(note)])
+    }
+
+    func testRecursivelyReceiveValue() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let publisher = makePublisher(center, for: name, object: nil)
+        let tracking = TrackingSubscriberBase<Notification, Never>(
+            receiveSubscription: { $0.request(.max(3)) },
+            receiveValue: { _ in .unlimited }
+        )
+        publisher.subscribe(tracking)
+
+        let note = Notification(name: name)
+        var recursionCounter = 7
+        tracking.onValue = { _ in
+            if recursionCounter == 0 { return }
+            recursionCounter -= 1
+            center.post(note)
+        }
+
+        center.post(note)
+
+        XCTAssertEqual(tracking.history, [.subscription("NotificationCenter Observer"),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note)])
+
+        center.post(note)
+
+        XCTAssertEqual(tracking.history, [.subscription("NotificationCenter Observer"),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note),
+                                          .value(note)])
+    }
+
+    func testCancelAlreadyCancelled() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let publisher = makePublisher(center, for: name, object: nil)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriberBase<Notification, Never>(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+        publisher.subscribe(tracking)
+
+        try XCTUnwrap(downstreamSubscription).cancel()
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        XCTAssertEqual(tracking.history, [.subscription("NotificationCenter Observer")])
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .removeObserver,
+                        .removeObserverForName(nil, nil)])
+    }
+
+    func testCancellingReleasesNotificationCenter() throws {
+        var centerDestroyed = false
+        var downstreamSubscription: Subscription?
+        do {
+            let center = TestNotificationCenter()
+            center.onDeinit = { centerDestroyed = true }
+            let name = Notification.Name(rawValue: "testName")
+            let publisher = makePublisher(center, for: name, object: nil)
+            let tracking = TrackingSubscriberBase<Notification, Never>(
+                receiveSubscription: { downstreamSubscription = $0 }
+            )
+            publisher.subscribe(tracking)
+        }
+        XCTAssertFalse(centerDestroyed)
+        try XCTUnwrap(downstreamSubscription).cancel()
+        XCTAssertTrue(centerDestroyed)
+    }
+
+    func testWeakCaptureWhenAddingObserver() {
+        let center = TestNotificationCenter()
+        let name = Notification.Name("testName")
+        var value: Notification?
+        do {
+            let publisher = makePublisher(center, for: name, object: nil)
+            let tracking = TrackingSubscriberBase<Notification, Never>(
+                receiveSubscription: { $0.request(.max(1)) },
+                receiveValue: { value = $0; return .none }
+            )
+            publisher.subscribe(tracking)
+            tracking.clearHistory() // Release the subscription
+        }
+        center.post(name: name, object: nil)
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil),
+                        .postNotificationWithName(name, nil, nil),
+                        .postNotification(Notification(name: name))])
+        XCTAssertNil(value)
+    }
+
+    func testZeroDemand() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let publisher = makePublisher(center, for: name, object: nil)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriberBase<Notification, Never>(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+        publisher.subscribe(tracking)
+        try XCTUnwrap(downstreamSubscription).request(.none)
+
+        XCTAssertEqual(center.history,
+                       [.addObserver(name, nil, nil)])
+        XCTAssertEqual(tracking.history, [.subscription("NotificationCenter Observer")])
+    }
+
+    func testNotificationCenterSubscriptionReflection() throws {
+        let center = TestNotificationCenter()
+        let name = Notification.Name(rawValue: "testName")
+        let object = TestObject.one
+        let publisher = makePublisher(center, for: name, object: object)
+
+        try testSubscriptionReflection(
+            description: "NotificationCenter Observer",
+            customMirror: expectedChildren(
+                ("center", .matches(String(describing: Optional(center)))),
+                ("name", .contains(String(describing: name))),
+                ("object", .matches(String(describing: Optional(object)))),
+                ("demand", "max(0)")
+            ),
+            playgroundDescription: "NotificationCenter Observer",
+            sut: publisher
+        )
+    }
+
+    func testEquatable() {
+        let center1 = NotificationCenter()
+        let center2 = NotificationCenter()
+        let name1 = Notification.Name(rawValue: "abcdefg")
+        let name2 = Notification.Name(rawValue: "1234567")
+        let object1 = TestObject.one
+        let object2 = TestObject.two
+
+        XCTAssertEqual(makePublisher(center1, for: name1, object: object1),
+                       makePublisher(center1, for: name1, object: object1))
+        XCTAssertEqual(makePublisher(center2, for: name2, object: object2),
+                       makePublisher(center2, for: name2, object: object2))
+        XCTAssertEqual(makePublisher(center1, for: name1, object: nil),
+                       makePublisher(center1, for: name1, object: nil))
+        XCTAssertNotEqual(makePublisher(center1, for: name1, object: object1),
+                          makePublisher(center1, for: name1, object: nil))
+        XCTAssertNotEqual(makePublisher(center1, for: name1, object: nil),
+                          makePublisher(center1, for: name1, object: object2))
+        XCTAssertNotEqual(makePublisher(center1, for: name1, object: object1),
+                          makePublisher(center1, for: name1, object: object2))
+        XCTAssertNotEqual(makePublisher(center1, for: name1, object: object1),
+                          makePublisher(center1, for: name2, object: object1))
+        XCTAssertNotEqual(makePublisher(center1, for: name1, object: object1),
+                          makePublisher(center2, for: name1, object: object1))
+    }
+}
+
+#if OPENCOMBINE_COMPATIBILITY_TEST || !canImport(Combine)
+@available(macOS 10.15, iOS 13.0, *)
+private func makePublisher(
+    _ center: NotificationCenter,
+    for name: Notification.Name,
+    object: AnyObject?
+) -> NotificationCenter.Publisher {
+    return center.publisher(for: name, object: object)
+}
+#else
+private func makePublisher(
+    _ center: NotificationCenter,
+    for name: Notification.Name,
+    object: AnyObject?
+) -> NotificationCenter.OCombine.Publisher {
+    return center.ocombine.publisher(for: name, object: object)
+}
+#endif
+
+/// A simple mock notification center that always sends notifications to **all**
+/// observers in non-thread safe manner.
+private final class TestNotificationCenter: NotificationCenter {
+
+    enum Event {
+        case postNotificationWithName(Notification.Name, Any?, [AnyHashable : Any]?)
+        case postNotification(Notification)
+        case addObserver(Notification.Name?, Any?, OperationQueue?)
+        case removeObserver
+        case removeObserverForName(Notification.Name?, Any?)
+    }
+
+    private final class Observation {
+        let callback: (Notification) -> Void
+
+        init(callback: @escaping (Notification) -> Void) {
+            self.callback = callback
+        }
+    }
+
+    private final class Token: NSObject {
+        weak var observation: Observation?
+
+        init(observer: TestNotificationCenter.Observation) {
+            self.observation = observer
+        }
+    }
+
+    private(set) var history = [Event]()
+
+    private var observations: [Observation] = []
+
+    var onDeinit: (() -> Void)?
+
+    deinit {
+        onDeinit?()
+    }
+
+    override func post(name aName: Notification.Name,
+                       object anObject: Any?,
+                       userInfo aUserInfo: [AnyHashable : Any]? = nil) {
+        history.append(.postNotificationWithName(aName, anObject, aUserInfo))
+        let notification = Notification(name: aName,
+                                        object: anObject,
+                                        userInfo: aUserInfo)
+        post(notification)
+    }
+
+    override func post(_ notification: Notification) {
+        history.append(.postNotification(notification))
+        for observation in observations {
+            observation.callback(notification)
+        }
+    }
+
+    override func addObserver(
+        forName name: NSNotification.Name?,
+        object obj: Any?,
+        queue: OperationQueue?,
+        using block: @escaping (Notification) -> Void
+    ) -> NSObjectProtocol {
+        history.append(.addObserver(name, obj, queue))
+        let observer = Observation(callback: block)
+        observations.append(observer)
+        return Token(observer: observer)
+    }
+
+    override func removeObserver(_ observer: Any) {
+        history.append(.removeObserver)
+        removeObserver(observer, name: nil, object: nil)
+    }
+
+    override func removeObserver(_ observer: Any,
+                                 name aName: NSNotification.Name?,
+                                 object anObject: Any?) {
+        history.append(.removeObserverForName(aName, anObject))
+        guard let observer = observer as? Token else { return }
+        observations.removeAll { $0 === observer.observation }
+    }
+}
+
+private final class TestObject: NSObject {
+
+    static let one = TestObject()
+
+    static let two = TestObject()
+}
+
+extension TestNotificationCenter.Event: Equatable {
+    fileprivate static func == (lhs: TestNotificationCenter.Event,
+                                rhs: TestNotificationCenter.Event) -> Bool {
+        switch (lhs, rhs) {
+        case let (.postNotification(lhsNote), .postNotification(rhsNote)):
+            return lhsNote == rhsNote
+        case let (.postNotificationWithName(lhsName,
+                                            lhsObject as TestObject?,
+                                            lhsUserInfo),
+                  .postNotificationWithName(rhsName,
+                                            rhsObject as TestObject?,
+                                            rhsUserInfo)):
+            return lhsName == rhsName &&
+                   lhsObject === rhsObject &&
+                   (lhsUserInfo == nil) == (rhsUserInfo == nil)
+        case let (.addObserver(lhsName, lhsObject as TestObject?, lhsQueue),
+                  .addObserver(rhsName, rhsObject as TestObject?, rhsQueue)):
+            return lhsName == rhsName &&
+                   lhsObject === rhsObject &&
+                   lhsQueue == rhsQueue
+        case (.removeObserver, .removeObserver):
+            return true
+        case let (.removeObserverForName(lhsName, lhsObject as TestObject?),
+                  .removeObserverForName(rhsName, rhsObject as TestObject?)):
+            return lhsName == rhsName && lhsObject === rhsObject
+        default:
+            return false
+        }
+    }
+}
+
+extension TestNotificationCenter.Event: CustomStringConvertible {
+    var description: String {
+        switch self {
+        case let .postNotificationWithName(name, object, userInfo):
+            return """
+            .postNotificationWithName(\
+            .init(rawValue: \"\(name.rawValue)\"), \
+            \(object.map(String.init(describing:)) ?? "nil"), \
+            \(userInfo.map(String.init(describing:)) ?? "nil"))
+            """
+        case .postNotification:
+            return ".postNotification(note)"
+        case let .addObserver(name, object, queue):
+            let nameDescription = name.map { ".init(rawValue: \($0.rawValue))" } ?? "nil"
+            return """
+            .addObserver(\
+            \(nameDescription), \
+            \(object.map(String.init(describing:)) ?? "nil"), \
+            \(queue.map(String.init(describing:)) ?? "nil"))
+            """
+        case .removeObserver:
+            return ".removeObserver"
+        case let .removeObserverForName(name, object):
+            let nameDescription = name.map { ".init(rawValue: \($0.rawValue))" } ?? "nil"
+            return """
+            .removeObserverForName(\
+            \(nameDescription), \
+            \(object.map(String.init(describing:)) ?? "nil"))
+            """
+        }
+    }
+}
diff --git a/Tests/OpenCombineTests/FoundationTests/PropertyListDecoderTests.swift b/Tests/OpenCombineTests/FoundationTests/PropertyListDecoderTests.swift
new file mode 100644
index 0000000..b83e492
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/PropertyListDecoderTests.swift
@@ -0,0 +1,76 @@
+//
+//  PropertyListDecoderTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class PropertyListDecoderTests: XCTestCase {
+    func testSuccessfullyDecode() {
+        let decoder = PropertyListDecoder()
+        let input = """
+        <?xml version="1.0" encoding="UTF-8"?>
+        <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" \
+        "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+        <plist version="1.0">
+        <array>
+        \t<dict>
+        \t\t<key>success</key>
+        \t\t<true/>
+        \t</dict>
+        </array>
+        </plist>
+
+        """
+        var actualOutput: [Subscribers.Completion<TestingError>]?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .map { Data($0.utf8) }
+            .decode(type: [Subscribers.Completion<TestingError>].self, decoder: decoder)
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+        switch actualCompletion {
+        case .finished?:
+            XCTAssertEqual(actualOutput, [.finished])
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+
+    func testDecodingFailure() {
+        let decoder = PropertyListDecoder()
+        let input = "000000"
+        var actualOutput: [Int]?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .map { Data($0.utf8) }
+            .decode(type: [Int].self, decoder: decoder)
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+        switch actualCompletion {
+        case .finished?:
+            XCTFail("Unexpected success")
+        case .failure(DecodingError.typeMismatch)?:
+            XCTAssertNil(actualOutput)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+}
diff --git a/Tests/OpenCombineTests/FoundationTests/PropertyListEncoderTests.swift b/Tests/OpenCombineTests/FoundationTests/PropertyListEncoderTests.swift
new file mode 100644
index 0000000..5bce561
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/PropertyListEncoderTests.swift
@@ -0,0 +1,81 @@
+//
+//  PropertyListEncoderTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 10.12.2019.
+//
+
+import Foundation
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class PropertyListEncoderTests: XCTestCase {
+
+    func testSuccessfullyEncode() {
+        let encoder = PropertyListEncoder()
+        encoder.outputFormat = .xml
+        let input = [Subscribers.Completion<TestingError>.finished]
+        var actualOutput: String?
+        var actualCompletion: Subscribers.Completion<Error>?
+
+        let cancellable = Just(input)
+            .encode(encoder: encoder)
+            .map { String(decoding: $0, as: UTF8.self) }
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+
+        switch actualCompletion {
+        case .finished?:
+            XCTAssertEqual(actualOutput, """
+            <?xml version="1.0" encoding="UTF-8"?>
+            <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" \
+            "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+            <plist version="1.0">
+            <array>
+            \t<dict>
+            \t\t<key>success</key>
+            \t\t<true/>
+            \t</dict>
+            </array>
+            </plist>
+
+            """)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+
+    func testEncodingFailure() {
+        let encoder = PropertyListEncoder()
+        let input = Double.nan
+        var actualOutput: String?
+        var actualCompletion: Subscribers.Completion<Error>?
+        let cancellable = Just(input)
+            .encode(encoder: encoder)
+            .map { String(decoding: $0, as: UTF8.self) }
+            .sink(receiveCompletion: { actualCompletion = $0 },
+                  receiveValue: { actualOutput = $0 })
+
+        switch actualCompletion {
+        case .finished?:
+            XCTFail("Unexpected success")
+        case .failure(EncodingError.invalidValue)?:
+            XCTAssertNil(actualOutput)
+        case .failure(let error)?:
+            XCTFail("Unexpected failure received: \(error)")
+        case nil:
+            XCTFail("Expected completion")
+        }
+        cancellable.cancel()
+    }
+}
diff --git a/Tests/OpenCombineTests/FoundationTests/URLSessionTests.swift b/Tests/OpenCombineTests/FoundationTests/URLSessionTests.swift
new file mode 100644
index 0000000..8d9789c
--- /dev/null
+++ b/Tests/OpenCombineTests/FoundationTests/URLSessionTests.swift
@@ -0,0 +1,724 @@
+//
+//  URLSessionTests.swift
+//  
+//
+//  Created by Sergej Jaskiewicz on 13.12.2019.
+//
+
+// swiftlint:disable multiline_arguments
+
+import Foundation
+import XCTest
+
+#if canImport(FoundationNetworking)
+import FoundationNetworking
+#endif
+
+// We can't test it on non-Darwin platforms because swift-corelibs-foundation
+// doesn't allow us to override some URLSession methods that we need.
+//
+// As soon as https://github.com/apple/swift-corelibs-foundation/pull/2587 makes it
+// into a release, we can enable these tests on non-Darwin platforms.
+//
+// The publisher itself though should work alright on those platforms.
+#if canImport(Darwin)
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+import OpenCombineFoundation
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class URLSessionTests: XCTestCase {
+
+    private typealias TrackingSubscriber =
+        TrackingSubscriberBase<(data: Data, response: URLResponse), URLError>
+
+    private let testURL = URL(string: "https://github.com")!
+
+    private let testRequest = URLRequest(url: URL(string: "https://github.com")!,
+                                         cachePolicy: .reloadIgnoringCacheData,
+                                         timeoutInterval: 42)
+
+    private let testData = Data("test data".utf8)
+
+    private let testResponse = URLResponse(url: URL(string: "https://example.com")!,
+                                           mimeType: "text/markdown",
+                                           expectedContentLength: 300,
+                                           textEncodingName: "utf-8")
+
+    private let testError = URLError(.cannotParseResponse, userInfo: ["a" : 1])
+
+    private let unknownError = URLError(.unknown)
+
+    func testDataTaskPublisherFromURL() {
+        let publisher = makePublisher(TestURLSession(testDataTask: .init()), testURL)
+        let expectedRequest = URLRequest(url: testURL)
+        XCTAssertEqual(publisher.request, expectedRequest)
+    }
+
+    func testDataTaskPublisherFromRequest() {
+        let publisher = makePublisher(TestURLSession(testDataTask: .init()), testRequest)
+        XCTAssertEqual(publisher.request, testRequest)
+    }
+
+    func testReceiveNothing() {
+        testReceiveResult(nil, nil, nil,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(unknownError))])
+    }
+
+    func testReceiveOnlyData() {
+        testReceiveResult(testData, nil, nil,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(unknownError))])
+    }
+
+    func testReceiveDataAndResponse() {
+        testReceiveResult(testData, testResponse, nil,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .value((testData, testResponse)),
+                                     .completion(.finished)])
+    }
+
+    func testReceiveDataAndURLError() {
+        testReceiveResult(testData, nil, testError,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(testError))])
+    }
+
+    func testReceiveDataAndUnrelatedError() {
+        testReceiveResult(testData, nil, TestingError.oops,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(unknownError))])
+    }
+
+    func testReceiveOnlyResponse() {
+        testReceiveResult(nil, testResponse, nil,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .value((Data(), testResponse)),
+                                     .completion(.finished)])
+    }
+
+    func testReceiveResponseAndURLError() {
+        testReceiveResult(nil, testResponse, testError,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(testError))])
+    }
+
+    func testReceiveResponseAndUnrelatedError() {
+        testReceiveResult(nil, testResponse, TestingError.oops,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(unknownError))])
+    }
+
+    func testReceiveOnlyURLError() {
+        testReceiveResult(nil, nil, testError,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(testError))])
+    }
+
+    func testReceiveOnlyUnrelatedError() {
+        testReceiveResult(nil, nil, TestingError.oops,
+                          expected: [.subscription("DataTaskPublisher"),
+                                     .completion(.failure(unknownError))])
+    }
+
+    func testRequesting() throws {
+        let dataTask = TestURLSessionDataTask()
+        let session = TestURLSession(testDataTask: dataTask)
+        let publisher = makePublisher(session, testRequest)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriber(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+        publisher.subscribe(tracking)
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [])
+        XCTAssertEqual(session.history, [])
+
+        session.completeDataTasks(testData, testResponse, nil)
+
+        try XCTUnwrap(downstreamSubscription).request(.max(2))
+        try XCTUnwrap(downstreamSubscription).request(.max(1))
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume, .resume])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume, .resume, .cancel])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+
+        session.completeDataTasks(testData, testResponse, nil)
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume, .resume, .cancel])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+    }
+
+    func testCancelAlreadyCancelled() throws {
+        let dataTask = TestURLSessionDataTask()
+        let session = TestURLSession(testDataTask: dataTask)
+        let publisher = makePublisher(session, testRequest)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriber(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+        publisher.subscribe(tracking)
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [])
+        XCTAssertEqual(session.history, [])
+
+        try XCTUnwrap(downstreamSubscription).request(.max(1))
+        try XCTUnwrap(downstreamSubscription).cancel()
+        try XCTUnwrap(downstreamSubscription).request(.max(1))
+        try XCTUnwrap(downstreamSubscription).cancel()
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume, .cancel])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+    }
+
+    func testCrashesOnZeroDemand() throws {
+        let dataTask = TestURLSessionDataTask()
+        let session = TestURLSession(testDataTask: dataTask)
+        let publisher = makePublisher(session, testURL)
+        var downstreamSubscription: Subscription?
+        let tracking = TrackingSubscriber(
+            receiveSubscription: { downstreamSubscription = $0 }
+        )
+        publisher.subscribe(tracking)
+
+        try assertCrashes {
+            try XCTUnwrap(downstreamSubscription).request(.none)
+        }
+    }
+
+    func testURLSessionSubscriptionReflection() throws {
+        let dataTask = TestURLSessionDataTask()
+        let session = TestURLSession(testDataTask: dataTask)
+        let publisher = makePublisher(session, testURL)
+        try testSubscriptionReflection(
+            description: "DataTaskPublisher",
+            customMirror: expectedChildren(
+                ("task", "nil"),
+                ("downstream", .contains("TrackingSubscriberBase")),
+                ("parent", .matches(String(describing: Optional(publisher)))),
+                ("demand", "max(0)")
+            ),
+            playgroundDescription: "DataTaskPublisher",
+            sut: publisher
+        )
+    }
+
+    // MARK: - Generic tests
+
+    private func testReceiveResult(_ data: Data?,
+                                   _ response: URLResponse?,
+                                   _ error: Error?,
+                                   expected: [TrackingSubscriber.Event]) {
+        let dataTask = TestURLSessionDataTask()
+        let session = TestURLSession(testDataTask: dataTask)
+        let publisher = makePublisher(session, testRequest)
+        let tracking = TrackingSubscriber(receiveSubscription: { $0.request(.max(1)) })
+        publisher.subscribe(tracking)
+
+        tracking.assertHistoryEqual([.subscription("DataTaskPublisher")],
+                                    valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+
+        session.completeDataTasks(data, response, error)
+        session.completeDataTasks(data, response, error)
+        session.completeDataTasks(data, response, error)
+
+        tracking.assertHistoryEqual(expected, valueComparator: ==)
+        XCTAssertEqual(dataTask.history, [.resume])
+        XCTAssertEqual(session.history, [.dataTaskWithRequestAndCompletion(testRequest)])
+    }
+}
+
+/// A simple mock URLSession that records its history and allows executing
+/// callbacks synchronously
+private class TestURLSession: URLSession {
+
+    enum Event: Equatable {
+        case delegateQueue
+        case delegate
+        case configuration
+        case getSessionDescription
+        case setSessionDescription(String?)
+        case finishTasksAndInvalidate
+        case invalidateAndCancel
+        case reset
+        case flush
+        case getTasksWithCompletionHandler
+        case getAllTasks
+        case dataTaskWithRequest(URLRequest)
+        case dataTaskWithRequestAndCompletion(URLRequest)
+        case dataTaskWithURL(URL)
+        case dataTaskWithURLAndCompletion(URL)
+        case uploadTaskWithRequestFromFile(URLRequest, URL)
+        case uploadTaskWithRequestFromFileWithCompletion(URLRequest, URL)
+        case uploadTaskWithRequestFromData(URLRequest, Data)
+        case uploadTaskWithRequestFromDataWithCompletion(URLRequest, Data?)
+        case uploadTaskWithStreamedRequest(URLRequest)
+        case downloadTaskWithRequest(URLRequest)
+        case downloadTaskWithRequestAndCompletion(URLRequest)
+        case downloadTaskWithURL(URL)
+        case downloadTaskWithURLAndCompletion(URL)
+        case downloadTaskWithResumeData(Data)
+        case downloadTaskWithResumeDataAndCompletion(Data)
+        case streamTaskWithHostNameAndPort(String, Int)
+#if canImport(Darwin) && swift(>=5.1)
+        case streamTaskWithService(NetService)
+        case webSocketTaskWithURL(URL)
+        case webSocketTaskWithURLAndProtocols(URL, [String])
+        case webSocketTaskWithRequest(URLRequest)
+#endif // canImport(Darwin) && swift(>=5.1)
+    }
+
+    private(set) var history = [Event]()
+
+    private(set) var dataTaskCompletionHandlers: [(Data?, URLResponse?, Error?) -> Void]
+
+    private let testDataTask: TestURLSessionDataTask
+
+    init(testDataTask: TestURLSessionDataTask) {
+        self.testDataTask = testDataTask
+        self.dataTaskCompletionHandlers = []
+    }
+
+    // MARK: Testing
+
+    func completeDataTasks(_ data: Data?, _ response: URLResponse?, _ error: Error?) {
+        for completionHandler in dataTaskCompletionHandlers {
+            completionHandler(data, response, error)
+        }
+    }
+
+    // MARK: Overrides
+
+    override class var shared: URLSession { fatalError("shared session is unavailable") }
+
+    override var delegateQueue: OperationQueue {
+        history.append(.delegateQueue)
+        return super.delegateQueue
+    }
+
+    override var delegate: URLSessionDelegate? {
+        history.append(.delegate)
+        return super.delegate
+    }
+
+    override var configuration: URLSessionConfiguration {
+        history.append(.configuration)
+        return super.configuration
+    }
+
+    override var sessionDescription: String? {
+        get {
+            history.append(.getSessionDescription)
+            return super.sessionDescription
+        }
+        set {
+            history.append(.setSessionDescription(newValue))
+            super.sessionDescription = newValue
+        }
+    }
+
+    override func finishTasksAndInvalidate() {
+        history.append(.finishTasksAndInvalidate)
+        super.finishTasksAndInvalidate()
+    }
+
+    override func invalidateAndCancel() {
+        history.append(.invalidateAndCancel)
+        super.invalidateAndCancel()
+    }
+
+    override func reset(completionHandler: @escaping () -> Void) {
+        history.append(.reset)
+        super.reset(completionHandler: completionHandler)
+    }
+
+    override func flush(completionHandler: @escaping () -> Void) {
+        history.append(.flush)
+        super.flush(completionHandler: completionHandler)
+    }
+
+    override func getTasksWithCompletionHandler(
+        _ completionHandler: @escaping ([URLSessionDataTask],
+                                        [URLSessionUploadTask],
+                                        [URLSessionDownloadTask]) -> Void
+    ) {
+        history.append(.getTasksWithCompletionHandler)
+        super.getTasksWithCompletionHandler(completionHandler)
+    }
+
+    @available(macOS 10.11, iOS 9.0, *)
+    override func getAllTasks(completionHandler: @escaping ([URLSessionTask]) -> Void) {
+        history.append(.getAllTasks)
+        super.getAllTasks(completionHandler: completionHandler)
+    }
+
+    override func dataTask(with request: URLRequest) -> URLSessionDataTask {
+        history.append(.dataTaskWithRequest(request))
+        return testDataTask
+    }
+
+    override func dataTask(with url: URL) -> URLSessionDataTask {
+        history.append(.dataTaskWithURL(url))
+        return testDataTask
+    }
+
+    override func dataTask(
+        with url: URL,
+        completionHandler: @escaping (Data?, URLResponse?, Error?) -> Void
+    ) -> URLSessionDataTask {
+        history.append(.dataTaskWithURLAndCompletion(url))
+        dataTaskCompletionHandlers.append(completionHandler)
+        return testDataTask
+    }
+
+    override func dataTask(
+        with request: URLRequest,
+        completionHandler: @escaping (Data?, URLResponse?, Error?) -> Void
+    ) -> URLSessionDataTask {
+        history.append(.dataTaskWithRequestAndCompletion(request))
+        dataTaskCompletionHandlers.append(completionHandler)
+        return testDataTask
+    }
+
+    override func uploadTask(with request: URLRequest,
+                             fromFile fileURL: URL) -> URLSessionUploadTask {
+        history.append(.uploadTaskWithRequestFromFile(request, fileURL))
+        return super.uploadTask(with: request, fromFile: fileURL)
+    }
+
+    override func uploadTask(with request: URLRequest,
+                             from bodyData: Data) -> URLSessionUploadTask {
+        history.append(.uploadTaskWithRequestFromData(request, bodyData))
+        return super.uploadTask(with: request, from: bodyData)
+    }
+
+    override func uploadTask(
+        withStreamedRequest request: URLRequest
+    ) -> URLSessionUploadTask {
+        history.append(.uploadTaskWithStreamedRequest(request))
+        return super.uploadTask(withStreamedRequest: request)
+    }
+
+    override func uploadTask(
+        with request: URLRequest,
+        fromFile fileURL: URL,
+        completionHandler: @escaping (Data?, URLResponse?, Error?) -> Void
+    ) -> URLSessionUploadTask {
+        history.append(.uploadTaskWithRequestFromFileWithCompletion(request, fileURL))
+        return super.uploadTask(with: request,
+                                fromFile: fileURL,
+                                completionHandler: completionHandler)
+    }
+
+    override func uploadTask(
+        with request: URLRequest,
+        from bodyData: Data?,
+        completionHandler: @escaping (Data?, URLResponse?, Error?) -> Void
+    ) -> URLSessionUploadTask {
+        history.append(.uploadTaskWithRequestFromDataWithCompletion(request, bodyData))
+        return super.uploadTask(with: request,
+                                from: bodyData,
+                                completionHandler: completionHandler)
+    }
+
+    override func downloadTask(with request: URLRequest) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithRequest(request))
+        return super.downloadTask(with: request)
+    }
+
+    override func downloadTask(with url: URL) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithURL(url))
+        return super.downloadTask(with: url)
+    }
+
+    override func downloadTask(
+        with request: URLRequest,
+        completionHandler: @escaping (URL?, URLResponse?, Error?) -> Void
+    ) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithRequestAndCompletion(request))
+        return super.downloadTask(with: request, completionHandler: completionHandler)
+    }
+
+    override func downloadTask(
+        with url: URL,
+        completionHandler: @escaping (URL?, URLResponse?, Error?) -> Void
+    ) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithURLAndCompletion(url))
+        return super.downloadTask(with: url, completionHandler: completionHandler)
+    }
+
+    override func downloadTask(
+        withResumeData resumeData: Data,
+        completionHandler: @escaping (URL?, URLResponse?, Error?) -> Void
+    ) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithResumeDataAndCompletion(resumeData))
+        return super.downloadTask(withResumeData: resumeData,
+                                  completionHandler: completionHandler)
+    }
+
+    override func downloadTask(
+        withResumeData resumeData: Data
+    ) -> URLSessionDownloadTask {
+        history.append(.downloadTaskWithResumeData(resumeData))
+        return super.downloadTask(withResumeData: resumeData)
+    }
+
+    @available(macOS 10.11, iOS 9.0, *)
+    override func streamTask(withHostName hostname: String,
+                             port: Int) -> URLSessionStreamTask {
+        history.append(.streamTaskWithHostNameAndPort(hostname, port))
+        return super.streamTask(withHostName: hostname, port: port)
+    }
+
+#if canImport(Darwin) && swift(>=5.1)
+    @available(macOS 10.11, iOS 9.0, *)
+    override func streamTask(with service: NetService) -> URLSessionStreamTask {
+        history.append(.streamTaskWithService(service))
+        return super.streamTask(with: service)
+    }
+
+    @available(macOS 10.15, iOS 13.0, *)
+    override func webSocketTask(with url: URL) -> URLSessionWebSocketTask {
+        history.append(.webSocketTaskWithURL(url))
+        return super.webSocketTask(with: url)
+    }
+
+    @available(macOS 10.15, iOS 13.0, *)
+    override func webSocketTask(with url: URL,
+                                protocols: [String]) -> URLSessionWebSocketTask {
+        history.append(.webSocketTaskWithURLAndProtocols(url, protocols))
+        return super.webSocketTask(with: url, protocols: protocols)
+    }
+
+    @available(macOS 10.15, iOS 13.0, *)
+    override func webSocketTask(with request: URLRequest) -> URLSessionWebSocketTask {
+        history.append(.webSocketTaskWithRequest(request))
+        return super.webSocketTask(with: request)
+    }
+#endif // canImport(Darwin) && swift(>=5.1)
+}
+
+private final class TestURLSessionDataTask: URLSessionDataTask {
+
+    enum Event: Equatable {
+        case taskIdentifier
+        case originalRequest
+        case currentRequest
+        case response
+        case progress
+        case getEarliestBeginDate
+        case setEarliestBeginDate(Date?)
+        case getCountOfBytesClientExpectsToSend
+        case setCountOfBytesClientExpectsToSend(Int64)
+        case getCountOfBytesClientExpectsToReceive
+        case setCountOfBytesClientExpectsToReceive(Int64)
+        case countOfBytesReceived
+        case countOfBytesSent
+        case countOfBytesExpectedToSend
+        case countOfBytesExpectedToReceive
+        case getTaskDescription
+        case setTaskDescription(String?)
+        case cancel
+        case state
+        case error
+        case suspend
+        case resume
+        case getPriority
+        case setPriority(Float)
+    }
+
+    private(set) var history = [Event]()
+
+    override init() {}
+
+    override var taskIdentifier: Int {
+        history.append(.taskIdentifier)
+        return super.taskIdentifier
+    }
+
+    override var originalRequest: URLRequest? {
+        history.append(.originalRequest)
+        return super.originalRequest
+    }
+
+    override var currentRequest: URLRequest? {
+        history.append(.currentRequest)
+        return super.currentRequest
+    }
+
+    override var response: URLResponse? {
+        history.append(.response)
+        return super.response
+    }
+
+    @available(macOS 10.13, iOS 11.0, *)
+    override var progress: Progress {
+        history.append(.progress)
+        return super.progress
+    }
+
+    @available(macOS 10.13, iOS 11.0, *)
+    override var earliestBeginDate: Date? {
+        get {
+            history.append(.getEarliestBeginDate)
+#if canImport(Darwin)
+            return super.earliestBeginDate
+#else
+            return nil // Deprecated in swift-corerlibs-foundation
+#endif
+        }
+        set {
+            history.append(.setEarliestBeginDate(newValue))
+#if canImport(Darwin)
+            super.earliestBeginDate = newValue
+#endif
+        }
+    }
+
+    @available(macOS 10.13, iOS 11.0, *)
+    override var countOfBytesClientExpectsToSend: Int64 {
+        get {
+            history.append(.getCountOfBytesClientExpectsToSend)
+            return super.countOfBytesClientExpectsToSend
+        }
+        set {
+            history.append(.setCountOfBytesClientExpectsToSend(newValue))
+            super.countOfBytesClientExpectsToSend = newValue
+        }
+    }
+
+    @available(macOS 10.13, iOS 11.0, *)
+    override var countOfBytesClientExpectsToReceive: Int64 {
+        get {
+            history.append(.getCountOfBytesClientExpectsToReceive)
+            return super.countOfBytesClientExpectsToReceive
+        }
+        set {
+            history.append(.setCountOfBytesClientExpectsToReceive(newValue))
+            super.countOfBytesClientExpectsToReceive = newValue
+        }
+    }
+
+    override var countOfBytesReceived: Int64 {
+        history.append(.countOfBytesReceived)
+        return super.countOfBytesReceived
+    }
+
+    override var countOfBytesSent: Int64 {
+        history.append(.countOfBytesSent)
+        return super.countOfBytesSent
+    }
+
+    override var countOfBytesExpectedToSend: Int64 {
+        history.append(.countOfBytesExpectedToSend)
+        return super.countOfBytesExpectedToSend
+    }
+
+    override var countOfBytesExpectedToReceive: Int64 {
+        history.append(.countOfBytesExpectedToReceive)
+        return super.countOfBytesExpectedToReceive
+    }
+
+    override var taskDescription: String? {
+        get {
+            history.append(.getTaskDescription)
+            return super.taskDescription
+        }
+        set {
+            history.append(.setTaskDescription(newValue))
+            super.taskDescription = newValue
+        }
+    }
+
+    override func cancel() {
+        history.append(.cancel)
+    }
+
+    override var state: URLSessionTask.State {
+        history.append(.state)
+        return super.state
+    }
+
+    override var error: Error? {
+        history.append(.error)
+        return super.error
+    }
+
+    override func suspend() {
+        history.append(.suspend)
+    }
+
+    override func resume() {
+        history.append(.resume)
+    }
+
+    override var priority: Float {
+        get {
+            history.append(.getPriority)
+            return super.priority
+        }
+        set {
+            history.append(.setPriority(newValue))
+            super.priority = newValue
+        }
+    }
+}
+
+extension URLError: EquatableError {}
+
+#if OPENCOMBINE_COMPATIBILITY_TEST || !canImport(Combine)
+@available(macOS 10.15, iOS 13.0, *)
+private func makePublisher(
+    _ session: URLSession,
+    _ url: URL
+) -> URLSession.DataTaskPublisher {
+    return session.dataTaskPublisher(for: url)
+}
+
+@available(macOS 10.15, iOS 13.0, *)
+private func makePublisher(
+    _ session: URLSession,
+    _ request: URLRequest
+) -> URLSession.DataTaskPublisher {
+    return session.dataTaskPublisher(for: request)
+}
+#else
+private func makePublisher(
+    _ session: URLSession,
+    _ url: URL
+) -> URLSession.OCombine.DataTaskPublisher {
+    return session.ocombine.dataTaskPublisher(for: url)
+}
+
+private func makePublisher(
+    _ session: URLSession,
+    _ request: URLRequest
+) -> URLSession.OCombine.DataTaskPublisher {
+    return session.ocombine.dataTaskPublisher(for: request)
+}
+#endif
+
+#endif // canImport(Darwin)
diff --git a/Tests/OpenCombineTests/Helpers/TestReflection.swift b/Tests/OpenCombineTests/Helpers/TestReflection.swift
index 5230a00..daca211 100644
--- a/Tests/OpenCombineTests/Helpers/TestReflection.swift
+++ b/Tests/OpenCombineTests/Helpers/TestReflection.swift
@@ -15,10 +15,10 @@ import OpenCombine
 
 let childrenIsEmpty: (Mirror) -> Bool = { $0.children.isEmpty }
 
-enum ExpectedMirrorChildValue: Equatable, ExpressibleByStringLiteral {
+enum ExpectedMirrorChildValue: ExpressibleByStringLiteral {
     case anything
-    case matches(String)
-    case contains(String)
+    case matches(@autoclosure () -> String)
+    case contains(@autoclosure () -> String)
 
     typealias StringLiteralType = String
 
@@ -49,10 +49,11 @@ func expectedChildren(_ expectedChildren: (String?, ExpectedMirrorChildValue)...
             case (_, .anything):
                 continue
             case let (lhs, .matches(rhs)):
-                XCTAssertEqual(lhs, rhs, file: file, line: line)
+                XCTAssertEqual(lhs, rhs(), file: file, line: line)
             case let (lhs, .contains(rhs)):
-                XCTAssert(lhs.contains(rhs),
-                          "\"\(lhs)\" doesn't contain substring \"\(rhs)\"",
+                let evaluatedRHS = rhs()
+                XCTAssert(lhs.contains(evaluatedRHS),
+                          "\"\(lhs)\" doesn't contain substring \"\(evaluatedRHS)\"",
                           file: file,
                           line: line)
             }
@@ -133,7 +134,7 @@ internal func testSubscriptionReflection<Sut: Publisher>(
     customMirror customMirrorPredicate: ((Mirror) -> Bool)?,
     playgroundDescription: String,
     sut: Sut
-) throws where Sut.Output: Equatable {
+) throws {
     let tracking = TrackingSubscriberBase<Sut.Output, Sut.Failure>()
     sut.subscribe(tracking)
 
diff --git a/Tests/OpenCombineTests/Helpers/TestingError.swift b/Tests/OpenCombineTests/Helpers/TestingError.swift
index fec9db9..f6c5c91 100644
--- a/Tests/OpenCombineTests/Helpers/TestingError.swift
+++ b/Tests/OpenCombineTests/Helpers/TestingError.swift
@@ -40,6 +40,20 @@ extension TestingError: ExpressibleByStringLiteral {
     }
 }
 
+protocol EquatableError: Error {
+    func isEqual(_ other: EquatableError) -> Bool
+}
+
+extension EquatableError where Self: Equatable {
+    func isEqual(_ other: EquatableError) -> Bool {
+        return self == (other as? Self)
+    }
+}
+
+extension TestingError: EquatableError {}
+
+extension NSError: EquatableError {}
+
 func assertThrowsError<Result>(_ expression: @autoclosure () throws -> Result,
                                _ expected: TestingError,
                                _ message: @autoclosure () -> String = "",
diff --git a/Tests/OpenCombineTests/Helpers/TrackingSubscriber.swift b/Tests/OpenCombineTests/Helpers/TrackingSubscriber.swift
index bc67db5..2040e32 100644
--- a/Tests/OpenCombineTests/Helpers/TrackingSubscriber.swift
+++ b/Tests/OpenCombineTests/Helpers/TrackingSubscriber.swift
@@ -5,6 +5,7 @@
 //  Created by Sergej Jaskiewicz on 11.06.2019.
 //
 
+import Foundation
 import XCTest
 
 #if OPENCOMBINE_COMPATIBILITY_TEST
@@ -177,6 +178,10 @@ final class TrackingSubscriberBase<Value, Failure: Error>
         for subscription in subscriptions {
             subscription.cancel()
         }
+        clearHistory()
+    }
+
+    func clearHistory() {
         history = []
     }
 
@@ -220,8 +225,8 @@ extension TrackingSubscriberBase.Event {
             switch (lhs, rhs) {
             case (.finished, .finished):
                 return true
-            case let (.failure(lhs), .failure(rhs)):
-                return (lhs as? TestingError) == (rhs as? TestingError)
+            case let (.failure(lhs as EquatableError), .failure(rhs as EquatableError)):
+                return lhs.isEqual(rhs)
             default:
                 return false
             }
@@ -277,8 +282,9 @@ final class TrackingSubjectBase<Output: Equatable, Failure: Error>
                 switch (lhs, rhs) {
                 case (.finished, .finished):
                     return true
-                case let (.failure(lhs), .failure(rhs)):
-                    return (lhs as? TestingError) == (rhs as? TestingError)
+                case let (.failure(lhs as EquatableError),
+                          .failure(rhs as EquatableError)):
+                    return lhs.isEqual(rhs)
                 default:
                     return false
                 }
diff --git a/Tests/OpenCombineTests/ObservableObjectPublisherTests.swift b/Tests/OpenCombineTests/ObservableObjectPublisherTests.swift
index 34ae2d3..27ac9fa 100644
--- a/Tests/OpenCombineTests/ObservableObjectPublisherTests.swift
+++ b/Tests/OpenCombineTests/ObservableObjectPublisherTests.swift
@@ -23,22 +23,27 @@ final class ObservableObjectPublisherTests: XCTestCase {
             receiveSubscription: { downstreamSubscription1 = $0 }
         )
         publisher.subscribe(tracking1)
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
         downstreamSubscription1?.request(.max(1))
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
         publisher.send()
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
                                       .signal])
         publisher.send()
         publisher.send()
         downstreamSubscription1?.request(.max(3))
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
+                                      .signal,
+                                      .signal,
                                       .signal])
         publisher.send()
         publisher.send()
         publisher.send()
         publisher.send()
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
+                                      .signal,
+                                      .signal,
+                                      .signal,
                                       .signal,
                                       .signal,
                                       .signal,
@@ -49,28 +54,48 @@ final class ObservableObjectPublisherTests: XCTestCase {
             receiveSubscription: { $0.request(.unlimited) }
         )
         publisher.subscribe(tracking2)
-        tracking2.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking2.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
 
         publisher.send()
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
+                                      .signal,
+                                      .signal,
+                                      .signal,
                                       .signal,
                                       .signal,
                                       .signal,
                                       .signal,
                                       .signal])
-        tracking2.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking2.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
                                       .signal])
 
         downstreamSubscription1?.cancel()
         publisher.send()
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
+                                      .signal,
+                                      .signal,
+                                      .signal,
                                       .signal,
                                       .signal,
                                       .signal,
                                       .signal,
                                       .signal])
-        tracking2.assertHistoryEqual([.subscription("PassthroughSubject"),
+        tracking2.assertHistoryEqual([.subscription("ObservableObjectPublisher"),
                                       .signal,
                                       .signal])
+
+        tracking1.cancel()
+        tracking2.cancel()
+    }
+
+    func testObservableObjectPublisherReflection() throws {
+        try testSubscriptionReflection(
+            description: "ObservableObjectPublisher",
+            customMirror: expectedChildren(
+                ("downstream", .contains("TrackingSubscriberBase"))
+            ),
+            playgroundDescription: "ObservableObjectPublisher",
+            sut: ObservableObjectPublisher()
+        )
     }
 }
diff --git a/Tests/OpenCombineTests/PublishedTests.swift b/Tests/OpenCombineTests/PublishedTests.swift
index e512428..ce5ce4f 100644
--- a/Tests/OpenCombineTests/PublishedTests.swift
+++ b/Tests/OpenCombineTests/PublishedTests.swift
@@ -88,11 +88,11 @@ final class PublishedTests: XCTestCase {
             receiveSubscription: { downstreamSubscription = $0 }
         )
         testObject.objectWillChange.subscribe(tracking1)
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
         downstreamSubscription?.request(.max(2))
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
         testObject.state = 100
-        tracking1.assertHistoryEqual([.subscription("PassthroughSubject")])
+        tracking1.assertHistoryEqual([.subscription("ObservableObjectPublisher")])
     }
 }
 
diff --git a/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift b/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
index b976888..5ea41ee 100644
--- a/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/ConcatenateTests.swift
@@ -359,9 +359,21 @@ final class ConcatenateTests: XCTestCase {
             helper.publisher.send(completion: .failure(.oops))
         }
 
-        assertCrashes {
-            helper.publisher.send(completion: .failure(.oops))
-        }
+        helper.publisher.send(completion: .failure(.oops))
+
+        XCTAssertEqual(helper.tracking.history, [.subscription("Concatenate"),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops)),
+                                                 .completion(.failure(.oops))])
+        XCTAssertEqual(helper.subscription.history, [])
     }
 
     func testHelperMethods() {
diff --git a/Tests/OpenCombineTests/PublisherTests/DelayTests.swift b/Tests/OpenCombineTests/PublisherTests/DelayTests.swift
index 71097b7..d5398ac 100644
--- a/Tests/OpenCombineTests/PublisherTests/DelayTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/DelayTests.swift
@@ -18,7 +18,7 @@ final class DelayTests: XCTestCase {
 
     // Delay's Inner doesn't conform to CustomStringConvertible, so we can't compare
     // subscriptions using their descriptions
-    let delaySubscription: StringSubscription = {
+    private let delaySubscription: StringSubscription = {
         let tracking = TrackingSubscriber()
         let scheduler = VirtualTimeScheduler()
         CustomPublisher(subscription: CustomSubscription())
@@ -29,6 +29,15 @@ final class DelayTests: XCTestCase {
             ?? "Delay"
     }()
 
+    private let delaySubscriptionImmediateScheduler: StringSubscription = {
+        let tracking = TrackingSubscriber()
+        CustomPublisher(subscription: CustomSubscription())
+            .delay(for: 0, scheduler: ImmediateScheduler.shared)
+            .subscribe(tracking)
+        return tracking.subscriptions.first.map(StringSubscription.subscription)
+            ?? "Delay"
+    }()
+
     func testBasicBehavior() {
         let scheduler = VirtualTimeScheduler()
         let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
@@ -41,15 +50,15 @@ final class DelayTests: XCTestCase {
         }
         XCTAssertNotNil(helper.publisher.subscriber)
 
-        XCTAssertEqual(helper.tracking.history, [])
-        XCTAssertEqual(helper.subscription.history, [])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
+        XCTAssertEqual(scheduler.history, [])
 
         scheduler.executeScheduledActions()
 
         XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
         XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(scheduler.history, [])
 
         XCTAssertEqual(helper.publisher.send(1), .none)
         XCTAssertEqual(helper.publisher.send(2), .none)
@@ -62,8 +71,7 @@ final class DelayTests: XCTestCase {
                                                   .nanoseconds(200)])
 
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(200),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions),
@@ -89,8 +97,7 @@ final class DelayTests: XCTestCase {
                                                      .requested(.max(12))])
 
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(200),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions),
@@ -117,8 +124,7 @@ final class DelayTests: XCTestCase {
                                                      .requested(.max(12))])
         XCTAssertEqual(scheduler.scheduledDates, [.nanoseconds(400)])
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(200),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions),
@@ -146,8 +152,7 @@ final class DelayTests: XCTestCase {
                                                      .requested(.max(12)),
                                                      .requested(.max(12))])
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(200),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions),
@@ -223,14 +228,14 @@ final class DelayTests: XCTestCase {
 
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
         XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(scheduler.history, [])
 
         XCTAssertEqual(helper.publisher.send(0), .none)
         helper.publisher.send(completion: .finished)
 
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
         XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(scheduler.history, [])
     }
 
     func testReceiveCompletionImmediatelyAfterSubscription() {
@@ -246,19 +251,19 @@ final class DelayTests: XCTestCase {
 
         helper.publisher.send(completion: .failure(.oops))
 
-        XCTAssertEqual(helper.tracking.history, [])
-        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited)])
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(123),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions)])
 
         scheduler.executeScheduledActions()
 
-        XCTAssertEqual(helper.tracking.history, [.completion(.failure(.oops))])
-        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(helper.tracking.history, [.subscription(delaySubscription),
+                                                 .completion(.failure(.oops))])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited)])
     }
 
     func testReceiveCompletionImmediatelyAfterValue() {
@@ -282,8 +287,7 @@ final class DelayTests: XCTestCase {
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
                                                      .requested(.max(418))])
         XCTAssertEqual(scheduler.history,
-                       [.schedule(options: .nontrivialOptions),
-                        .now,
+                       [.now,
                         .scheduleAfterDate(.nanoseconds(123),
                                            tolerance: .nanoseconds(5),
                                            options: .nontrivialOptions),
@@ -313,13 +317,29 @@ final class DelayTests: XCTestCase {
             $0.delay(for: .nanoseconds(123), scheduler: ImmediateScheduler.shared)
         }
 
+        var recursionCounter = 5
         helper.tracking.onValue = { _ in
+            if recursionCounter == 0 { return }
+            recursionCounter -= 1
             _ = helper.publisher.send(-1)
         }
 
-        assertCrashes {
-            _ = helper.publisher.send(0)
-        }
+        XCTAssertEqual(helper.publisher.send(0), .none)
+        XCTAssertEqual(helper.tracking.history,
+                       [.subscription(delaySubscriptionImmediateScheduler),
+                        .value(0),
+                        .value(-1),
+                        .value(-1),
+                        .value(-1),
+                        .value(-1),
+                        .value(-1)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418))])
     }
 
     func testReceiveCompletionRecursively() {
@@ -334,27 +354,7 @@ final class DelayTests: XCTestCase {
         helper.publisher.send(completion: .finished)
     }
 
-    func testWeakCaptureWhenSchedulingSubscription() {
-        let scheduler = VirtualTimeScheduler()
-        var subscription: Subscription?
-        var subscriberReleased = false
-        do {
-            let publisher = CustomPublisher(subscription: CustomSubscription())
-            let delay = publisher.delay(for: 0.35, scheduler: scheduler)
-            let tracking = TrackingSubscriber(receiveSubscription: { subscription = $0 },
-                                              onDeinit: { subscriberReleased = true })
-            delay.subscribe(tracking)
-            XCTAssertEqual(tracking.history, [])
-            XCTAssertEqual(scheduler.history, [.minimumTolerance,
-                                               .schedule(options: nil)])
-            publisher.cancel()
-        }
-        XCTAssertTrue(subscriberReleased)
-        scheduler.executeScheduledActions()
-        XCTAssertNil(subscription)
-    }
-
-    func testWeakCaptureWhenSchedulingValue() {
+    func testStrongCaptureWhenSchedulingValue() {
         let scheduler = VirtualTimeScheduler()
         var value: Int?
         var subscriberReleased = false
@@ -370,7 +370,6 @@ final class DelayTests: XCTestCase {
             XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
             XCTAssertEqual(scheduler.history,
                            [.minimumTolerance,
-                            .schedule(options: nil),
                             .now,
                             .scheduleAfterDate(.seconds(0.35),
                                                tolerance: 0,
@@ -380,11 +379,11 @@ final class DelayTests: XCTestCase {
         }
         XCTAssertFalse(subscriberReleased)
         scheduler.executeScheduledActions()
-        XCTAssertNil(value)
+        XCTAssertEqual(value, 42)
         XCTAssertTrue(subscriberReleased)
     }
 
-    func testWeakCaptureWhenSchedulingCompletion() {
+    func testStrongCaptureWhenSchedulingCompletion() {
         let scheduler = VirtualTimeScheduler()
         var completion: Subscribers.Completion<TestingError>?
         var subscriberReleased = false
@@ -400,7 +399,6 @@ final class DelayTests: XCTestCase {
             XCTAssertEqual(tracking.history, [.subscription(delaySubscription)])
             XCTAssertEqual(scheduler.history,
                            [.minimumTolerance,
-                            .schedule(options: nil),
                             .now,
                             .scheduleAfterDate(.seconds(0.35),
                                                tolerance: 0,
@@ -410,7 +408,7 @@ final class DelayTests: XCTestCase {
         }
         XCTAssertFalse(subscriberReleased)
         scheduler.executeScheduledActions()
-        XCTAssertNil(completion)
+        XCTAssertEqual(completion, .finished)
         XCTAssertTrue(subscriberReleased)
     }
 
diff --git a/Tests/OpenCombineTests/PublisherTests/FlatMapTests.swift b/Tests/OpenCombineTests/PublisherTests/FlatMapTests.swift
index 102e439..b649a3b 100644
--- a/Tests/OpenCombineTests/PublisherTests/FlatMapTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/FlatMapTests.swift
@@ -390,11 +390,15 @@ final class FlatMapTests: XCTestCase {
     }
 
     func testChildValueReceivedWhileSendingValue() throws {
-        let upstreamPublisher = PassthroughSubject<AnyPublisher<Int, TestingError>,
-            TestingError>()
+        let upstreamSubscription = CustomSubscription()
+        let upstreamPublisher = CustomPublisherBase<CustomPublisher, TestingError>(
+            subscription: upstreamSubscription
+        )
 
-        let child1Publisher = CustomPublisher(subscription: CustomSubscription())
-        let child2Publisher = CustomPublisher(subscription: CustomSubscription())
+        let childSubscription1 = CustomSubscription()
+        let childSubscription2 = CustomSubscription()
+        let child1Publisher = CustomPublisher(subscription: childSubscription1)
+        let child2Publisher = CustomPublisher(subscription: childSubscription2)
 
         let flatMap = upstreamPublisher.flatMap { $0 }
 
@@ -408,12 +412,17 @@ final class FlatMapTests: XCTestCase {
 
         flatMap.subscribe(downstreamSubscriber)
 
-        upstreamPublisher.send(AnyPublisher(child1Publisher))
-        upstreamPublisher.send(AnyPublisher(child2Publisher))
+        XCTAssertEqual(upstreamPublisher.send(child1Publisher), .none)
+        XCTAssertEqual(upstreamPublisher.send(child2Publisher), .none)
 
-        assertCrashes {
-            XCTAssertEqual(child1Publisher.send(666), .max(1))
-        }
+        XCTAssertEqual(child1Publisher.send(666), .max(1))
+
+        XCTAssertEqual(upstreamSubscription.history, [.requested(.unlimited)])
+        XCTAssertEqual(downstreamSubscriber.history, [.subscription("FlatMap"),
+                                                      .value(666),
+                                                      .value(777)])
+        XCTAssertEqual(childSubscription1.history, [.requested(.max(1))])
+        XCTAssertEqual(childSubscription2.history, [.requested(.max(1))])
     }
 
     func testOuterLockReentrance() {
@@ -462,9 +471,6 @@ final class FlatMapTests: XCTestCase {
         // Create some downstream demand
         try XCTUnwrap(helper.downstreamSubscription).request(.max(5))
 
-        // If Apple changes the implementation to use recursive lock,
-        // we must make sure no stack overflow occurs here,
-        // which will also be detected as a crash, which is not what we want.
         var recursionDepth = 10
         helper.tracking.onFailure = { _ in
             if recursionDepth <= 0 {
@@ -474,10 +480,13 @@ final class FlatMapTests: XCTestCase {
             _ = child.send(1)
         }
 
-        // Expected deadlock
-        assertCrashes {
-            child.send(completion: .failure(.oops))
-        }
+        child.send(completion: .failure(.oops))
+
+        XCTAssertEqual(helper.tracking.history, [.subscription("FlatMap"),
+                                                 .completion(.failure(.oops)),
+                                                 .value(1)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(1))])
+        XCTAssertEqual(childSubscription.history, [.requested(.max(1))])
     }
 
     func testCompletesProperlyWhenUpstreamOutlivesChildren() {
diff --git a/Tests/OpenCombineTests/PublisherTests/MapErrorTests.swift b/Tests/OpenCombineTests/PublisherTests/MapErrorTests.swift
index 6c0d12a..afb340a 100644
--- a/Tests/OpenCombineTests/PublisherTests/MapErrorTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/MapErrorTests.swift
@@ -210,10 +210,15 @@ final class MapErrorTests: XCTestCase {
     }
 }
 
-private struct OtherError: Error {
-    let original: Error
+private struct OtherError: EquatableError {
+    let original: EquatableError
 
-    init(_ original: Error) {
+    init(_ original: EquatableError) {
         self.original = original
     }
+
+    func isEqual(_ other: EquatableError) -> Bool {
+        guard let other = other as? OtherError else { return false }
+        return original.isEqual(other.original)
+    }
 }
diff --git a/Tests/OpenCombineTests/PublisherTests/ReceiveOnTests.swift b/Tests/OpenCombineTests/PublisherTests/ReceiveOnTests.swift
index 49fa41b..56825d0 100644
--- a/Tests/OpenCombineTests/PublisherTests/ReceiveOnTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/ReceiveOnTests.swift
@@ -26,15 +26,15 @@ final class ReceiveOnTests: XCTestCase {
         XCTAssertNotNil(helper.publisher.subscriber,
                         "Subscription must be performed synchronously")
 
-        XCTAssertEqual(helper.tracking.history, [])
-        XCTAssertEqual(helper.subscription.history, [])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn")])
+        XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
+        XCTAssertEqual(scheduler.history, [])
 
         scheduler.executeScheduledActions()
 
         XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn")])
         XCTAssertEqual(helper.subscription.history, [.requested(.max(100))])
-        XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions)])
+        XCTAssertEqual(scheduler.history, [])
 
         XCTAssertEqual(helper.publisher.send(1), .none)
         XCTAssertEqual(helper.publisher.send(2), .none)
@@ -47,7 +47,6 @@ final class ReceiveOnTests: XCTestCase {
                                                   .nanoseconds(0)])
 
         XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions),
-                                           .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions)])
 
@@ -64,7 +63,6 @@ final class ReceiveOnTests: XCTestCase {
                                                      .requested(.max(12))])
 
         XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions),
-                                           .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions)])
 
@@ -82,7 +80,6 @@ final class ReceiveOnTests: XCTestCase {
                                                      .requested(.max(12))])
         XCTAssertEqual(scheduler.scheduledDates, [.nanoseconds(0)])
         XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions),
-                                           .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions)])
@@ -97,7 +94,6 @@ final class ReceiveOnTests: XCTestCase {
                                                      .requested(.max(12)),
                                                      .requested(.max(12))])
         XCTAssertEqual(scheduler.history, [.schedule(options: .nontrivialOptions),
-                                           .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions),
                                            .schedule(options: .nontrivialOptions)])
@@ -155,14 +151,14 @@ final class ReceiveOnTests: XCTestCase {
 
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
         XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn")])
-        XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
+        XCTAssertEqual(scheduler.history, [])
 
         XCTAssertEqual(helper.publisher.send(0), .none)
         helper.publisher.send(completion: .finished)
 
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited), .cancelled])
         XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn")])
-        XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
+        XCTAssertEqual(scheduler.history, [])
     }
 
     func testReceiveCompletionImmediatelyAfterSubscription() {
@@ -175,15 +171,15 @@ final class ReceiveOnTests: XCTestCase {
 
         helper.publisher.send(completion: .failure(.oops))
 
-        XCTAssertEqual(helper.tracking.history, [])
-        XCTAssertEqual(helper.subscription.history, [])
-        XCTAssertEqual(scheduler.history, [.schedule(options: nil),
-                                           .schedule(options: nil)])
+        XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn")])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited)])
+        XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
 
         scheduler.executeScheduledActions()
 
-        XCTAssertEqual(helper.tracking.history, [.completion(.failure(.oops))])
-        XCTAssertEqual(helper.subscription.history, [])
+        XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn"),
+                                                 .completion(.failure(.oops))])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited)])
     }
 
     func testReceiveCompletionImmediatelyAfterValue() {
@@ -204,7 +200,6 @@ final class ReceiveOnTests: XCTestCase {
         XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
                                                      .requested(.max(418))])
         XCTAssertEqual(scheduler.history, [.schedule(options: nil),
-                                           .schedule(options: nil),
                                            .schedule(options: nil),
                                            .schedule(options: nil)])
 
@@ -218,20 +213,35 @@ final class ReceiveOnTests: XCTestCase {
                                                      .requested(.max(418))])
     }
 
-    func testCrashesWhenReceivingInputRecursively() {
+    func testReceiveInputRecursively() {
         let helper = OperatorTestHelper(publisherType: CustomPublisher.self,
                                         initialDemand: .unlimited,
                                         receiveValueDemand: .max(418)) {
             $0.receive(on: ImmediateScheduler.shared)
         }
 
+        var recursionCounter = 5
         helper.tracking.onValue = { _ in
+            if recursionCounter == 0 { return }
+            recursionCounter -= 1
             _ = helper.publisher.send(-1)
         }
 
-        assertCrashes {
-            _ = helper.publisher.send(0)
-        }
+        XCTAssertEqual(helper.publisher.send(0), .none)
+        XCTAssertEqual(helper.tracking.history, [.subscription("ReceiveOn"),
+                                                 .value(0),
+                                                 .value(-1),
+                                                 .value(-1),
+                                                 .value(-1),
+                                                 .value(-1),
+                                                 .value(-1)])
+        XCTAssertEqual(helper.subscription.history, [.requested(.unlimited),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418)),
+                                                     .requested(.max(418))])
     }
 
     func testReceiveCompletionRecursively() {
@@ -246,25 +256,6 @@ final class ReceiveOnTests: XCTestCase {
         helper.publisher.send(completion: .finished)
     }
 
-    func testWeakCaptureWhenSchedulingSubscription() {
-        let scheduler = VirtualTimeScheduler()
-        var subscription: Subscription?
-        var subscriberReleased = false
-        do {
-            let publisher = CustomPublisher(subscription: CustomSubscription())
-            let receiveOn = publisher.receive(on: scheduler)
-            let tracking = TrackingSubscriber(receiveSubscription: { subscription = $0 },
-                                              onDeinit: { subscriberReleased = true })
-            receiveOn.subscribe(tracking)
-            XCTAssertEqual(tracking.history, [])
-            XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
-            publisher.cancel()
-        }
-        XCTAssertTrue(subscriberReleased)
-        scheduler.executeScheduledActions()
-        XCTAssertNil(subscription)
-    }
-
     func testWeakCaptureWhenSchedulingValue() {
         let scheduler = VirtualTimeScheduler()
         var value: Int?
@@ -279,8 +270,7 @@ final class ReceiveOnTests: XCTestCase {
             XCTAssertEqual(tracking.history, [.subscription("ReceiveOn")])
             XCTAssertEqual(publisher.send(42), .none)
             XCTAssertEqual(tracking.history, [.subscription("ReceiveOn")])
-            XCTAssertEqual(scheduler.history, [.schedule(options: nil),
-                                               .schedule(options: nil)])
+            XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
             tracking.cancel()
             publisher.cancel()
         }
@@ -304,8 +294,7 @@ final class ReceiveOnTests: XCTestCase {
             XCTAssertEqual(tracking.history, [.subscription("ReceiveOn")])
             publisher.send(completion: .finished)
             XCTAssertEqual(tracking.history, [.subscription("ReceiveOn")])
-            XCTAssertEqual(scheduler.history, [.schedule(options: nil),
-                                               .schedule(options: nil)])
+            XCTAssertEqual(scheduler.history, [.schedule(options: nil)])
             tracking.cancel()
             publisher.cancel()
         }
diff --git a/Tests/OpenCombineTests/PublisherTests/SubscribeOnTests.swift b/Tests/OpenCombineTests/PublisherTests/SubscribeOnTests.swift
index c1332e6..e6f5745 100644
--- a/Tests/OpenCombineTests/PublisherTests/SubscribeOnTests.swift
+++ b/Tests/OpenCombineTests/PublisherTests/SubscribeOnTests.swift
@@ -206,7 +206,10 @@ final class SubscribeOnTests: XCTestCase {
             $0.subscribe(on: ImmediateScheduler.shared)
         }
 
+        var recursionCounter = 5
         helper.subscription.onRequest = { _ in
+            if recursionCounter == 0 { return }
+            recursionCounter -= 1
             helper.downstreamSubscription?.request(.unlimited)
         }