Rxswift: Discussion: using scan to implement a running diff

[AndrewSB]

There are two ways that I've seen the scan operator used in codebases.

One is for a moving value, where there the value returned is like an accumulator. Modeling a moving average, for example

Observable.scan(MovingAverage()) { average, newValue in return (average * 0.5) + (newValue * 0.5) }

scan affords itself to this use case really well...

The second use case that I've often seen scan used for isn't as elegant - when used to diff a new element from a stream with the previous, and use that diff to perform work. For example,

Observable.scan([SessionType]()) { [weak self] old, new in
    let sessionsThatAreNoLongerAround = old.filter {
        oldSession in new.contains { $0 == oldSession }
    }
    // do something (that can't be asynchronous) with the sessionsThatAreNoLongerAround
    // or onNext(sessionsThatAreNoLongerAround) to a Subject

    return new
}.flatMap { 
    // you don't have access to the diff here 😰, just the new `.next` element from the Observable 
}

observes a source that publishes the current Sessions, and then uses the figures out which sessions have dropped off, to update some UI, or send a network request.
The problem with using scan for this usage, is that it becomes completely uncomposable.

scan is forced to return the new element, to have the .next(element) diffed correctly. So either a new observable has to be created within the scan closure, or another Subject must be created.

I've written something along the lines of scanMap:

func scanMap(_ seed: E, diff: @escaping (E, E) -> E) -> SharedSequence<SharingStrategy, E> {

    var diffed: E! = nil
    return self
            .scan(seed, accumulator: { (oldValue, newValue) in
                diffed = diff(oldValue, newValue)

                return newValue
            })
            .map { _ in diffed! }
}

Which lets you solve the second use case without shelling out to another subject, or creating an Observable within an Observable without flat mapping it.

If this operator makes sense, I think it would be a really useful addition. If theres a better way to solve the problem, I'd love to talk about it 😄

[kzaher]

Hi @AndrewSB ,

I'll try to answer questions one by one.

I'm not sure I understand. It seem to me that solution for this is simply:

Observable.scan([SessionType]()) { [weak self] old, new -> (newValue, diff) in

}.flatMapLatest { (newValue, diff) in
    // you do have access to the diff here and the new `.next` element from the Observable 
}

I also have a couple of scan convenience operators that I define in some of my projects.

The general rule is that we don't add convenience operators in this library because we could literally add thousands of those.

We try to add as orthogonal operators as possible to avoid this issue.

The following code is not a valid operator code because diffed is shared between multiple subscriptions.

func scanMap(_ seed: E, diff: @escaping (E, E) -> E) -> SharedSequence<SharingStrategy, E> {
    var diffed: E! = nil
    return self
            .scan(seed, accumulator: { (oldValue, newValue) in
                diffed = diff(oldValue, newValue)

                return newValue
            })
            .map { _ in diffed! }
}

If you really want to solve it in similar way, you could theoretically do the following, but I strongly advise against it because it is unnecessary creating sideeffects and explicitly unwrapping optionals.

func scanMap(_ seed: E, diff: @escaping (E, E) -> E) -> SharedSequence<SharingStrategy, E> {
   return Observable.deferred {
        var diffed: E! = nil
        return self
            .scan(seed, accumulator: { (oldValue, newValue) in
                diffed = diff(oldValue, newValue)

                return newValue
            })
            .map { _ in diffed! }
    }
}

I would suggest my first code snippet instead.

[AndrewSB]

Ahh, I see how my diffed variable will be shared between multiple subscribers - thats definitely a problem with my implementation. I've currently switched to yours with the Observable.deferred.

I'd really like to use your first snippet, but I'm having trouble understanding it. The signature of scan is scan(seed: A, accumulator: (A, Element) -> A), your snippet initially defines A to be [SessionType], as the seed, but then you return a tuple (newValue, diff), which definitely isn't of type [SessionType] in the accumulator.

Can you help me understand how the types work out? Should you start the seed with a tuple instead?

[sergdort]

Hi, guys
Sorry if my question is out of the scope of the topic:)
Just woundering @kzaher if you could some how share your custom operators, at least some of that you think are solving common problems 🙂, may be in RxSwiftExt
Because, from my experience people tend to solve problems imperatively if they struggle to find solution with Rx

[kzaher]

Hi guys,

@AndrewSB yeah, sorry, it should be something like

let initial: ([SessionType], somekindofdifftype) = ([], [])
Observable.scan() { [weak self] old, new -> (newValue, diff) in

}.flatMapLatest { (newValue, diff) -> what ever is needed in
    // you do have access to the diff here and the new `.next` element from the Observable 
}

Code example was meant to be just an illustration. If there is no sensible default value for diff, then I guess defining let initial: AccumulatorType where AccumulatorType is some enum that has initial case defined might make sense.

@sergdort
yeah, I'm trying out some ideas. I would be happy to share them publicly when I polish them a bit and test them in real world scenarios.

[AndrewSB]

@kzaher that looks perfect, updating my project to use that instead. Much thanks 😄

[AndrewSB]

@kzaher I ended up writing a small helper, can you give it a quick look over and let me know if you see any problems with it?
It's identical to the solution you provided, it just maps the final result to remove the latest element, and just give the diffed value

/**
     RunningDiff simplifies the operation of maintaining a running diff from an observable value.

     @param seed: The initial value to diff against
     @param diffSeed: The initial diff value
     @param diff: A function that is able to compute the diff between two Observable elements
     */
    // swiftlint:disable:next variable_name
    func runningDiff<A>(seed: E, diffSeed: A, diff: @escaping (E, E) -> A) -> Observable<A> {
        return self
            .scan((seed, diffSeed), accumulator: { running, new -> (E, A) in
                let diffed = diff(running.0, new)

                return (new, diffed)
            })
            .map { _, diffed in diffed }
    }

Here's an example of how it works

Observable
    .from(["i", "hi", "hello"])
    .runningDiff(seed: "", diffSeed: 0, diff: { old, new in new.characters.count - old.characters.count })
    .subscribe(onNext: { print($0) })

prints 1                         1                         3
       ^ (i.count - "".count)    ^ (hi.count - i.count)    ^ (hello.count - hi.count)

[kzaher]

I don't see any issues with it.

👍

[ghost]

Best solution

pod repo remove master
git clone https://github.com/CocoaPods/Specs.git ~/.cocoapods/repos/master
pod setup