Experimenting with flatMap, concat and buffer in RxJava.

zip-vs-from.groovy

@Grab('com.netflix.rxjava:rxjava-joins:0.19.6')
import rx.Observable
import rx.functions.Func1
import rx.functions.Action1

import java.util.concurrent.TimeUnit

def range = new IntRange(5, 0)
range.each { println it}

println "from and toList"

List<Integer> numbers = Observable.from(range).flatMap({ i ->
    return Observable.just(i).delay(i, TimeUnit.SECONDS).doOnNext({ n ->
        println "Done with ${n}"
    } as Action1)
} as Func1).toList().toBlocking().first()

println "Done with ${numbers}"

println "from, concat, toList"

numbers = Observable.concat(Observable.from(range).map({ i ->
    return Observable.just(i).delay(i, TimeUnit.SECONDS).doOnNext({ n ->
        println "Done with ${n}"
    } as Action1)
} as Func1)).toList().toBlocking().first()

println "Done with ${numbers}"

println "buffer 2"

numbers = Observable.from(range).buffer(2).flatMap({ window ->
    // Taking first item from window as delay
    int delay = window[0]
    return Observable.just(window).delay(delay, TimeUnit.SECONDS).doOnNext({ n ->
        println "Done with ${n}"
    } as Action1)
} as Func1).toList().toBlocking().first()

println "Done with ${numbers}"

println "buffer 2 with concat"

numbers = Observable.concat(Observable.from(range).buffer(2).map({ window ->
    // Taking first item from window as delay
    int delay = window[0]
    return Observable.just(window).delay(delay, TimeUnit.SECONDS).doOnNext({ n ->
        println "Done with ${n}"
    } as Action1)
} as Func1)).toList().toBlocking().first()

println "Done with ${numbers}"