KISS

Keep It Simple Stupid

Sequencing multiple SignalProducers

| comments

Let’s say we have a list of usernames and an API that can return the details of a user by her username, and we need to make the requests one after another. We will use ReactiveSwift as a framework to work with asynchronous values. This task involves working with multiple, although the same-typed SignalProducers. The code, step by step, is available at https://github.com/eunikolsky/SignalProducerExtensions.

Note: the code here will be the essence necessary for understanding of the post, without the swift’s boilerplate. The links for the full files are provided.

The domain model

UserDetails.swift has the very simple domain model:

1
2
3
4
5
public typealias Username = String

public struct UserDetails {
    let name: Username
}

And we need to implement the UserDetailsFetcher class that will return an array of UserDetails for an array of Usernames:

1
2
3
4
5
6
7
8
9
10
11
12
13
public struct UserDetailsFetcher {
    /// A type of function to make an API request to get `UserDetails`.
    /// This will send only one value and complete. For simplicity, we don't
    /// handle errors.
    public typealias NetworkRequest = (Username) -> SignalProducer<UserDetails, Never>

    let networkRequest: NetworkRequest

    /// Sequentially fetches details for every `Username`.
    public func fetchDetails(for usernames: [Username]) -> SignalProducer<[UserDetails], Never> {
        return .empty
    }
}

NetworkRequest represents an interface to asynchronously get the user’s details by her username, which is typically implemented by making a real network request. But it’s too boring and unreliable in our tests, so they can now inject a more suitable version. Here’s a simple UserDetailsFetcherSpec for what we need to do:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
describe("fetchDetails") {
    it("returns details for all users") {
        let fakeNetworkRequest = { username in
            SignalProducer(value: UserDetails(name: "~" + username))
                .delay(0.001, on: QueueScheduler(qos: .userInteractive))
        }

        let sut = UserDetailsFetcher(networkRequest: fakeNetworkRequest)
        let detailsProducer = sut.fetchDetails(for: ["foo", "bar", "!"])
        let result = detailsProducer.single()

        let expected = [UserDetails(name: "~foo"), UserDetails(name: "~bar"), UserDetails(name: "~!")]
        let actual = try! result?.get()
        expect(actual).to(equal(expected))
    }
}

The test of course fails now.

Primitive implementation

An easy-to-write, hard-to-understand, imperative solution could be this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public func fetchDetails(for usernames: [Username]) -> SignalProducer<[UserDetails], Never> {
    return SignalProducer { observer, lifetime in
        var restOfUsernames = usernames
        var allDetails = [UserDetails]()

        func iteration() {
            if restOfUsernames.isEmpty {
                observer.send(value: allDetails)
                observer.sendCompleted()
                return
            }

            let username = restOfUsernames.removeFirst()
            self.networkRequest(username)
                .on(completed: {
                    iteration()
                },
                    value: { details in
                        allDetails.append(details)
                    })
                .start()
        }

        iteration()
    }
}

Here we have the iteration function that handles every username from a mutable copy of the input array: it starts the networkRequest, adds a value to the mutable allDetails when it’s received, and calls itself when the request is completed. The next iteration will have the first username removed. If we don’t have usernames anymore, we send the accumulated result and exit. All this happens only when the SignalProducer returned from fetchDetails is started.

It works, but there are multiple issues with the code:

  • The business logic (calling networkRequest) is mixed with the SignalProducer machinery (sending the values, observing the values and completion event).

  • It uses two mutable variables, which makes it harder to track down when they are mutated.

  • You can’t interrupt the inner requests by disposing of the main signal producer (lifetime is not used). There is a retain cycle here, which resolves itself when all the producers complete, but not before that.

  • This explicit recursion could be useful to other similar functions, but it is now hardcoded in this function.

Implementation using sequence

We can do better than that. An important step here is to abstract the recursive collection of the results from the SignalProducers. What if we try to map every username to the request function?

1
2
3
public func fetchDetails(for usernames: [Username]) -> SignalProducer<[UserDetails], Never> {
    return usernames.map(self.fetcher)
}

We get the error: Cannot convert return expression of type '[SignalProducer<UserDetails, Never>]' to return type 'SignalProducer<[UserDetails], Never>', but look at the types: basically we have Array<SignalProducer> and we need SignalProducer<Array> — we just need to swap the two layers! I haven’t found a function to do that in the ReactiveSwift library. If we extract this function, it could be made generic and help us remove a lot of the SignalProducer machinery from fetchDetails. It’s not hard to implement it:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public extension Collection where Element: SignalProducerProtocol, Element.Error == Never {
    /// Sequentially executes an array of `SignalProducer`s and collects the results in an array.
    /// The type is effectively is:
    /// `sequence : Array<SignalProducer<T, E>> -> SignalProducer<Array<T>, E>`
    func sequence() -> SignalProducer<[Element.Value], Element.Error> {
        func iter <C: Collection> (_ producers: C, _ results: [Element.Value]) -> SignalProducer<[Element.Value], Element.Error> where C.Element == Element {
            switch producers.first {
            case .some(let producer):
                return producer.producer.flatMap(.merge, { value in
                    iter(producers.dropFirst(), results + [value])
                })

            case .none:
                return SignalProducer(value: results)
            }
        }

        return iter(self, [])
    }
}

It also has an iterative iter function inside, which receives the rest of the producers and the results so far. If we can get the first element (the collection is not empty), we chain a lambda to the current producer with flatMap, where that lambda will call another iteration with one fewer producer and one more result. Here the flatMap function does the sequencing of two signal producers. If there are no more producers, return a signal producers that will emit the accumulated result.

Now our implementation is very simple indeed:

1
2
3
4
5
public func fetchDetails(for usernames: [Username]) -> SignalProducer<[UserDetails], Never> {
    return usernames
        .map(self.networkRequest)
        .sequence()
}

Only the business logic left!

Extra notes

The flatMap combinator does the internal sequencing, which gives us a few benefits, such as:

  • If we dispose of the returned SignalProducer, the current inner producer will be interrupted and the whole chain will stop as expected.

  • There is absolutely no need for the Element.Error == Never requirement in the sequence() definition above. If you remove it, you can easily work with failable signal producers, such that if an error is encountered at any step, it will be returned as the result of the whole chain, and the rest won’t be processed.

Why the sequence name? There is a function with this name in Haskell, which does conceptually the same thing, but is much more generic than is possible in swift:

1
sequence :: (Traversable t, Monad m) => t (m a) -> m (t a)

Traversable is a “data structure that can be traversed from left to right”, Collection in the implementation above. Monad, briefly, represents a way to compose functions with effects, the SignalProvider is a Monad in our code (flatMap is a method defined for any Monad). Substituting the more concrete types, we could get:

1
sequence :: Collection (SignalProducer a) -> SignalProducer (Collection a)

Comments