Create publisher extension that can be executed on every event

Question

I'm having hard time learning combine.

What I try to reach is to create some helper for mockedLoader that will fail from time to time.

I wrote such extension:

struct AnyError: Error {}

extension Publisher where Failure == Error {
    func fail(withProbability probability: Float, error: Failure = AnyError()) -> AnyPublisher<Output, Failure> {
        let treshold = Int.random(in: 0..<100)
        let p = Int(probability * 100)
        if treshold + p >= 100 {
            return Fail(error: error).eraseToAnyPublisher()
        }
        return self.eraseToAnyPublisher()
    }
}

When I'm trying to use it in my Publisher like that:

let mockedLoader: AnyPublisher<[TransactionItem], Error> = client.getPublisher(url: URL(string: "https://testing.com")!)
     .subscribe(on: upstreamQueue)
     .fail(withProbability: 0.05)
     .delay(for: .seconds(2), tolerance: .milliseconds(500), scheduler: upstreamQueue)
     .tryMap(TransactionsMapper.map)
     .receive(on: DispatchQueue.main)
     .eraseToAnyPublisher()

I realized the fail block is executed only once at the beginning when Publisher is created. I would like to have this probability logic called every time there is a event and based on logic return Fail of self publisher.

What I'm doing wrong here. Should I use some operator but which one?

Answer 1

It's probably easier to add a TryMap operator into the chain instead trying to create a custom Publisher:

First, implement this function:

func fail<Value>(
    value: Value, 
    with probability: Float, 
    error: Error
) throws -> Value {
    ...
}

Then, in your chain of Combine operators, insert a tryMap where the upstream is the value which may also fail:

    someOperatorPublishingValue
    .tryMap { value in 
        try fail(value: value, with: 0.1, error: MyError())
    } 
    ...

You may choose Value to be what suites your needs. In your case, it's likely the output of the client.getPublisher publisher.

You may need to adjust for error types in the downstream.