CombineでObservable.create/SignalProducer.initに相当するPublisherを作る

Combine には当初、クロージャを通じて複数の値の出力と成否を送る Publisher (RxSwift における Observable.create または ReactiveSwift における SignalProducer.init) が実装されていたようなのですが、途中で削除されてしまったらしい...
どうしても使いたい場面があったため、Combineの学習がてら自分で実装してみました。

こちらの投稿を参照にさせていただきました🙇‍♂️
https://qiita.com/shiz/items/58abf44b77d9da2042f1

SomePublisher.swift

import Foundation
import Combine

// 名前は適当です...
struct SomePublisher<Output, Failure: Swift.Error>: Publisher {
    enum Event {
        case value(Output)
        case finished
        case failure(Failure)
    }
    typealias Handler = (Event) -> Void
    private let handler: (@escaping Handler) -> Void

    private class Subscription<S: Subscriber>: Combine.Subscription where S.Input == Output, S.Failure == Failure {
        private var subscriber: S?
        private let handler: (@escaping Handler) -> Void

        init(subscriber: S, handler: @escaping (@escaping Handler) -> Void) {
            self.subscriber = subscriber
            self.handler = handler
        }

        func request(_ demand: Subscribers.Demand) {
            self.handler(self.handle)
        }

        func cancel() {
            self.subscriber = nil
        }

        private func handle(_ event: Event) {
            guard let subscriber = self.subscriber else {
                return
            }

            switch event {
            case .value(let input):
                _ = subscriber.receive(input)
            case .finished:
                subscriber.receive(completion: .finished)
            case .failure(let error):
                subscriber.receive(completion: .failure(error))
            }
        }
    }

    init(_ handler: @escaping (@escaping Handler) -> Void) {
        self.handler = handler
    }

    func receive<S>(subscriber: S) where S : Subscriber, S.Failure == Failure, S.Input == Output {
        let subscription = Subscription(subscriber: subscriber, handler: self.handler)
        subscriber.receive(subscription: subscription)
    }
}

使い方はこんな感じ。

func values(count: Int) -> SomePublisher<Int, Never> {
    return .init { (publisher) in
        // ここにストリームを実装する...
        for i in 0..<count {
            publisher(.value(i))
        }
        publisher(.finished)
    }
}

本当はイニシャライザを以下のような仕様にしたかったのですが、 Subscriber の管理と受け渡しがうまく実装できなかったので、 Event 型を使ってストリームする実装にしています。

    init<S>(_ handler: @escaping (S) -> Void) where S : Subscriber, S.Failure == Failure, S.Input == Output {
        ...
    }

Combineに実装が復活するといいですね。