Example of cancellable custom publisher

api_example.swift

extension Service {
    private func request(_ request: RequestConvertible) -> Request {
        self.resource(request.path)
            .request(request.method) { urlRequest in
                request.headers?.forEach { urlRequest.addValue($0.value, forHTTPHeaderField: $0.key) }
                request.mutation?(&urlRequest)
            }
    }

    struct RequestPublisher: Publisher {
        typealias Output = Void
        typealias Failure = Error

        private let service: Service
        private let requestModel: RequestConvertible

        init(service: Service, request: RequestConvertible) {
            self.service = service
            self.requestModel = request
        }

        func receive<S>(subscriber: S) where S: Subscriber, Failure == S.Failure, Output == S.Input {
            let request = service.request(requestModel)
                .onSuccess { response in
                    //notify about success
                    _ = subscriber.receive()
                    subscriber.receive(completion: .finished)
                }
                .onFailure { error in
                    //notify about failure
                    subscriber.receive(completion: .failure(error))
                }
                .start()
            subscriber.observe(cancel: { request.cancel() })
        }
    }

    func publisher(request: RequestConvertible) -> RequestPublisher {
        return .init(service: self, request: request)
    }
}

generic_subscription.swift

extension Subscriptions {
    struct Cancellable: Subscription {
        let combineIdentifier = CombineIdentifier()
        private let cancelBlock: () -> Void
        private let requestDemand: ((Subscribers.Demand) -> Void)?
        
        init(cancel: @escaping () -> Void, requestDemand: ((Subscribers.Demand) -> Void)? = nil) {
            cancelBlock = cancel
            self.requestDemand = requestDemand
        }
        
        func cancel() { cancelBlock() }
        func request(_ demand: Subscribers.Demand) { requestDemand?(demand)  }
    }
}

extension Subscriber {
    func observe(cancel: (() -> Void)? = nil, demand: ((Subscribers.Demand) -> Void)? = nil) {
        self.receive(subscription: Subscriptions.Cancellable(cancel: { cancel?() }, request: { demand?($0) }))
    }
}