Update .cancellable implementation to fix #946

[iampatbrown]

Possible fix for #946
Main issue with PassthroughSubject is that it could only forward 1 value from an upstream PassthroughSubject, ie. nested cancellable effects. This should fix that by subscribing directly to the upstream publisher.

[iampatbrown]

.recordButtonTapped ends up returning a merged effect

swift-composable-architecture/Examples/VoiceMemos/VoiceMemos/VoiceMemos.swift

Lines 70 to 76 in 313dd21

	return .merge(
	environment.audioRecorder.startRecording(url)
	.catchToEffect(VoiceMemosAction.audioRecorder),
	Effect.timer(id: TimerId(), every: 1, tolerance: .zero, on: environment.mainRunLoop)
	.map { _ in .currentRecordingTimerUpdated }
	)

using RunLoop.immediate results in a failed test because .currentRecordingTimerUpdated now also gets received once before .audioRecorder(.failure(.couldntActivateAudioSession)) is received. I'm unsure if this is an undesired result of the change.

Adding the following test to TimerTests.swift demonstrates the difference in behaviour.

 func testTimerImmediate() {
    let scheduler = DispatchQueue.immediate

    var count = 0

    Effect.timer(id: 1, every: .seconds(1), on: scheduler)
      .sink { _ in count += 1 }
      .store(in: &self.cancellables)

    XCTAssertNoDifference(count, 1) // This PR

    // XCTAssertNoDifference(count, 0) // Current implementation
  }

[iampatbrown]

For the current implementation, the timer is triggered as soon as the PassThroughSubject subscribes to it, essentially dropping the value before the effect is subscribed to downstream.

swift-composable-architecture/Sources/ComposableArchitecture/Effects/Cancellation.swift

Lines 36 to 37 in 313dd21

	let subject = PassthroughSubject<Output, Failure>()
	let cancellable = self.subscribe(subject)

So the timer auto connects and outputs a value at line 37 which is lost before returning the effect.

In this PR the timer auto connects at .sink { _ in count += 1 }

[stephencelis]

Interesting find! I think overall this PR is an improvement in the expectation that at least 1 thing emits from a timer running on an immediate scheduler, even if it is a breaking change.

[stephencelis]

Great work @iampatbrown! Thanks for digging into the problem.