CustomPublisher - customizable publisher for Combine framework

CustomPublisher.swift

import Combine

public final class CustomPublisher<Output, Failure>: Publisher where Failure: Error {

    public init(subscribe subscribeClosure: @escaping (AnySubscriber<Output, Failure>) -> Subscription) {
        self.subscribeClosure = subscribeClosure
    }

    public func receive<S>(subscriber: S) where S: Combine.Subscriber, S.Input == Output, S.Failure == Failure {
        let subscription = subscribeClosure(AnySubscriber(subscriber))
        subscriber.receive(subscription: subscription)
    }

    private let subscribeClosure: (AnySubscriber<Output, Failure>) -> Subscription

}

public extension CustomPublisher {

    convenience init(
        request requestClosure: @escaping (AnySubscriber<Output, Failure>, Subscribers.Demand) -> Void,
        cancel cancelClosure: @escaping () -> Void = {},
        deinit deinitClosure: @escaping () -> Void = {}
    ) {
        self.init { subscriber in
            CustomSubscription(
                subscriber,
                request: requestClosure,
                cancel: cancelClosure,
                deinit: deinitClosure
            )
        }
    }

}

public final class CustomSubscription<Output, Failure>: Subscription where Failure: Error {

    public init(
        _ subscriber: AnySubscriber<Output, Failure>,
        request requestClosure: @escaping (AnySubscriber<Output, Failure>, Subscribers.Demand) -> Void,
        cancel cancelClosure: @escaping () -> Void = {},
        deinit deinitClosure: @escaping () -> Void = {}
    ) {
        self.subscriber = subscriber
        self.requestClosure = requestClosure
        self.cancelClosure = cancelClosure
        self.deinitClosure = deinitClosure
    }

    deinit {
        deinitClosure()
    }

    public func request(_ demand: Subscribers.Demand) {
        if let subscriber = subscriber {
            requestClosure(subscriber, demand)
        }
    }

    public func cancel() {
        subscriber = nil
        cancelClosure()
    }

    private var subscriber: AnySubscriber<Output, Failure>?
    private let requestClosure: (AnySubscriber<Output, Failure>, Subscribers.Demand) -> Void
    private let cancelClosure: () -> Void
    private let deinitClosure: () -> Void

}

CustomPublisher_Example.swift

import Combine
import Foundation

func log(_ text: String) {
    let thread = Thread.isMainThread ? "[main thread] " : "[other thread]"
    print("\(thread) \(text)")
}

func createPublisher() -> AnyPublisher<String, Error> {
    CustomPublisher(subscribe: { subscriber in
        log("[PUBLISHER-SUBSCRIBE] subscriber: \(subscriber)")
        return CustomSubscription(subscriber, request: { subscriber, demand in
            log("[SUBSCRIPTION-REQUEST] subscriber: \(subscriber) demand: \(demand)")
            DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
                let value = "Text A"
                log("[SUBSCRIPTION-PRODUCE] value: \(value)")
                _ = subscriber.receive(value)
            }
            DispatchQueue.global().asyncAfter(deadline: .now() + 2) {
                let value = "Text B"
                log("[SUBSCRIPTION-PRODUCE] value: \(value)")
                _ = subscriber.receive(value)
            }
            DispatchQueue.global().asyncAfter(deadline: .now() + 3) {
                let completion = Subscribers.Completion<Error>.finished
                log("[SUBSCRIPTION-COMPLETE] completion: \(completion)")
                subscriber.receive(completion: .finished)
            }
        }, cancel: {
            log("[SUBSCRIPTION-CANCEL]")
        }, deinit: {
            log("[SUBSCRIPTION-DEINIT]")
        })
    }).eraseToAnyPublisher()
}

let subscription = createPublisher()
    .subscribe(on: DispatchQueue.global(qos: .background))
    .receive(on: DispatchQueue.main)
    .sink(receiveCompletion: { completion in
        log("[SINK-RECEIVED] completion: \(completion)")
    }, receiveValue: { value in
        log("[SINK-RECEIVED] value: \(value)")
    })

// Prints:
// -------
// [other thread] [PUBLISHER-SUBSCRIBE] subscriber: SubscribeOn
// [other thread] [SUBSCRIPTION-REQUEST] subscriber: SubscribeOn demand: unlimited
// [other thread] [SUBSCRIPTION-PRODUCE] value: Text A
// [main thread]  [SINK-RECEIVED] value: Text A
// [other thread] [SUBSCRIPTION-PRODUCE] value: Text B
// [main thread]  [SINK-RECEIVED] value: Text B
// [other thread] [SUBSCRIPTION-COMPLETE] completion: finished
// [other thread] [SUBSCRIPTION-DEINIT]
// [main thread]  [SINK-RECEIVED] completion: finished