Custom publisher acting up when upstream is subscribed by multiple subscribers

Biclops 0
Viewed 28

My custom publisher acts strangely if I have another subscriber subscribing to the upstream publisher.

So in my example, my upstream publisher gates the values dispatched

let onlyGreaterThan4 = pub.drop(while: { value in
    return value < 4
})

I have two subscribers, one with a normal sink and one in a custom publisher.

import UIKit
import Foundation
import Combine

var iter = 0
var pub: AnyPublisher<Int, Never> = Timer.publish(every: 1, on: .main, in: .common).autoconnect().map { _ -> Int in
    iter += 1
    print("sending \(iter)")
    return iter
}.eraseToAnyPublisher()

/// Collects values and starts a timer after the `first` value from upstream is received. This differs from the collect(.byTime) publisher since that publisher collects on an interval
public struct CollectOnReceive<T: Publisher>: Publisher where T.Failure == Never {
    public typealias Output = [T.Output]
    public typealias Failure = T.Failure

    /// Subscription for ReactiveSwiftPublisher
    class Subscription<SubscriberType: Subscriber>: Combine.Subscription where SubscriberType.Input == Output, SubscriberType.Failure == Failure {
        private var cancellables: Set<AnyCancellable> = Set()

        private let subscriber: SubscriberType
        private let upstream: T
        private let queue: DispatchQueue
        private var collection: Output = []

        init(upstream: T, subscriber: SubscriberType, queue: DispatchQueue) {
            self.subscriber = subscriber
            self.upstream = upstream
            self.queue = queue
        }

        func request(_ demand: Subscribers.Demand) {
            let subscriber = self.subscriber

            Swift.print("-- observing --")

            // collect all the values
            self.upstream.subscribe(on: self.queue)
                .sink { [weak self] value in
                    Swift.print("-- appending: \(value)")
                    self?.collection.append(value)
                }.store(in: &self.cancellables)

            // on the first item
            self.upstream.first()
                .flatMap({ _ -> AnyPublisher<Date, Never> in
                    // dvttodo using dispatch interval timer might be necessary
                    Swift.print("start: \(Date())")
                    return Timer.publish(every: 5, on: .main, in: .common).autoconnect().eraseToAnyPublisher()
                }).subscribe(on: self.queue)
                .sink(receiveValue: { [weak self] _ in
                    guard let collection = self?.collection else {
                        subscriber.receive(completion: .finished)
                        return
                    }

                    Swift.print("--finish sub: \(collection) : \(Date())")
                    subscriber.receive(collection)
                    subscriber.receive(completion: .finished)
                }).store(in: &self.cancellables)
        }

        func cancel() {
            self.cancellables.removeAll()
        }
    }

    private let upstream: T
    private let queue: DispatchQueue

    public init(upstream: T, queue: DispatchQueue) {
        self.upstream = upstream
        self.queue = queue
    }

    public func receive<S>(subscriber: S) where S : Subscriber, T.Failure == S.Failure, [T.Output] == S.Input {
        subscriber.receive(
            subscription: Subscription(
                upstream: self.upstream,
                subscriber: subscriber,
                queue: self.queue
            )
        )
    }
}

extension Publisher where Failure == Never {
    func collectOnReceive(queue: DispatchQueue = .init(label: "CollectOnReceive")) -> CollectOnReceive<Self> {
        return CollectOnReceive(upstream: self, queue: queue)
    }
}

let onlyGreaterThan4 = pub.drop(while: { value in
    return value < 4
})

let q = DispatchQueue(label: "ASDF")

let cancel = onlyGreaterThan4.collectOnReceive(queue: DispatchQueue(label: "collect on receive")).sink {
    print("completion: \($0)")
} receiveValue: {
    print("values: \($0)")
}

let c2 = onlyGreaterThan4.sink {
    print("c2 sink: \($0)")
}

// if you comment out c2, it acts mostly reasonable, the publisher completes properly and sends out the following array [4, 6, 7, 8, 9, 10, 11], don't know what happened to the 5 but that's mostly correct.

// if you leave second subscription with c2 in tact, then it returns the following: [6, 8, 10, 12, 14, 16]. Where did all the odd numbers go? No idea.

I figure I must be dealing with the dispatch queues wrong or something silly.

The reason I created a custom publisher was that I wanted value collection for an interval only after the first value greater than or equal to 4 was discovered. Using collect(.byTime) would not be suitable since it executes on the interval, regardless of when upstream values come into it.

So the pseudo code of the goal

when x >= 4 was detected then {
    collect(values, for: timeInterval) { valuesGreaterThanEqualTo4 in 
        if valuesGreaterThanEqualTo4.count > threshold {
            return sustained 
        }

        return notSustained
    }
}

Update: I noticed in my print that my sink subscription all my even values where going there while my odd values were going to my custom publisher. So weird.

-- observing --
sending 1
sending 2
sending 3
sending 4
c2 sink: 4
sending 5
-- appending: 5
sending 6
start: 2022-09-08 22:45:54 +0000
sending 7
-- appending: 7
sending 8
c2 sink: 8. // missing the append for all the even values!!!
sending 9
-- appending: 9
sending 10
c2 sink: 10
sending 11
-- appending: 11
sending 12
c2 sink: 12
sending 13
-- appending: 13
sending 14
c2 sink: 14
sending 15
-- appending: 15
sending 16
c2 sink: 16
--finish sub: [5, 7, 9, 11, 13, 15] : 2022-09-08 22:45:59 +0000
values: [5, 7, 9, 11, 13, 15]
publisher combine swift
1 Answers

I may not understand your problem statement, but let me try. See the playground below.

My integers is the same thing as your pub but without the global variable. Hiding it in the scan operator will keep anyone else from messing with it :-)

Next, you want all the integers greater than 4. So we create a signal for that with moreThan4. I'm going to use that as the base for more than one other signal that should all share the same value so I added the .share() operator.

Now you want the system to signal you when it first encounters a value greater than 4. Let's capture that explicitly with a signal. startCollecting will send a message when the first value greater than 4 is encountered.

When we get that message we want to transform it into a new signal. Transformation is the job of a map operation. In this case we want to transform the message into a new signal (instead of into a new value). The operator for turning values into new signals is flatMap.

The signal we want to transform the "start collecting" event into is one which reads from moreThan4 over time interval. And we only want to measure the first time interval. So we express that in code... collect from moreThan4 for 5 seconds and return the first such collection.

Is that what you were trying to accomplish?

import Foundation
import Combine

let integers = Timer.publish(every: 1.0,
                             on: RunLoop.main,
                             in: .common)
    .autoconnect()
    .scan(0) { count, _ in
        return count + 1
    }

let moreThan4 = integers
    .filter { $0 > 4 }
    .share()

let startCollecting = moreThan4.first()

let subscription = startCollecting
    .flatMap { _ in
        moreThan4
            .collect(.byTime(RunLoop.main, .seconds(5)))
            .first()
    }
    .sink { print($0) }
Scott Thompson0
Related