Transforming Observables

本文的主题为转换 Observable 的操作符。

这里的 Observable 实质上是可观察的数据流。

RxJava操作符（二）Transforming Observables

公共代码

• RxNET

public static void Dump<T>(this IObservable<T> source, string name)

{

    source.Subscribe(

    i => Console.WriteLine("{0}-->{1}", name, i),

    ex => Console.WriteLine("{0} failed-->{1}", name, ex.Message),

    () => Console.WriteLine("{0} completed", name));

}

• RxJava

fun <T> Observable<T>.dump() =

    this.subscribe({ println("onNext: $it") },

        { println("onError: $it: ${it.message}") },

        { println("onComplete") })

fun <T> Observable<T>.dump(name: String) =

    this.subscribe({ println("$name: onNext: $it") },

        { println("$name: onError: $it: ${it.message}") },

        { println("$name: onComplete") })

Buffer

ReactiveX - Buffer operator

Reactive Extensions再入門 その２５「値をまとめるBufferメソッド」

Buffer 转换数据流：间隔性地将数据流中邻近的数据打包，形成数据包的数据流。

这里的数据包即静态的数组。

• RxNET

var source = Observable.Interval(TimeSpan.FromSeconds(1)).Take(10);

source.Buffer(3, 1)

.Subscribe(

buffer =>

{

    Console.WriteLine("--Buffered values");

    foreach (var value in buffer)

        Console.WriteLine(value);

}, () => Console.WriteLine("Completed"));

/*

--Buffered values

0

1

2

--Buffered values

1

2

3

..

--Buffered values

7

8

9

--Buffered values

8

9

--Buffered values

9

Completed

*/

• RxJava

Observable.range(0, 10)

    .buffer(4)

    .dump()

/*

onNext: [0, 1, 2, 3]

onNext: [4, 5, 6, 7]

onNext: [8, 9]

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS).take(10)

    .buffer(250, TimeUnit.MILLISECONDS)

    .dump()

/*

onNext: [0, 1]

onNext: [2, 3, 4]

onNext: [5, 6]

onNext: [7, 8, 9]

onNext: []

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS)

    .take(10)

    .buffer(250, TimeUnit.MILLISECONDS, 2)

    .dump()

/*

onNext: [0, 1]

onNext: []

onNext: [2, 3]

onNext: []

onNext: [4, 5]

onNext: [6]

onNext: [7, 8]

onNext: [9]

onNext: []

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS).take(10)

    .buffer(Observable.interval(250, TimeUnit.MILLISECONDS))

    .dump()

/*

onNext: [0, 1]

onNext: [2, 3]

onNext: [4, 5, 6]

onNext: [7, 8, 9]

onNext: []

onComplete

*/

Observable.range(0, 10)

    .buffer(4, 3)

    .dump()

/*

onNext: [0, 1, 2, 3]

onNext: [3, 4, 5, 6]

onNext: [6, 7, 8, 9]

onNext: [9]

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS).take(10)

    .buffer(350, 200, TimeUnit.MILLISECONDS)

    .dump()

/*

onNext: [0, 1, 2]

onNext: [2, 3, 4]

onNext: [3, 4, 5, 6]

onNext: [6, 7, 8]

onNext: [7, 8, 9]

onNext: []

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS).take(10)

    .buffer<Long, Long>(

        Observable.interval(250, TimeUnit.MILLISECONDS),

        Function { i -> Observable.timer(200, TimeUnit.MILLISECONDS) } )

    .dump()

/*

onNext: [2, 3]

onNext: [5, 6]

onNext: [7, 8]

onNext: []

onComplete

*/

FlatMap / SelectMany

ReactiveX - FlatMap operator

Reactive Extensions再入門 その４０「IObservableの合成はじめました」

FlatMap / SelectMany 转换数据流：将源数据流的每一项都转换成数据流，从而形成数据流的数据流，最后再平坦化将两维数据流合并成一个数据流。

• RxNET

Observable.Range(1, 3)

.SelectMany(i => Observable.Range(1, i))

.Dump("SelectMany");

/*

SelectMany-->1

SelectMany-->1

SelectMany-->2

SelectMany-->1

SelectMany-->2

SelectMany-->3

SelectMany completed

*/

• RxJava

val values = Observable.just(2)

values

    .flatMap { i -> Observable.range(0, i) }

    .dump()

/*

onNext: 0

onNext: 1

onComplete

*/

val values = Observable.range(1, 3)

values

    .flatMap { i -> Observable.range(0, i) }

    .dump()

/*

onNext: 0

onNext: 0

onNext: 1

onNext: 0

onNext: 1

onNext: 2

onComplete

*/

val values = Observable.just(1)

values

    .flatMap { i ->

        Observable.just(

            Character.valueOf((i + 64).toChar())

        )

    }

    .dump()

/*

onNext: A

onComplete

*/

val values = Observable.range(0, 30)

values

    .flatMap<Char> { i ->

        if (i in 1..26)

            Observable.just(Character.valueOf((i + 64).toChar()))

        else

            Observable.empty()

    }

    .dump()

/*

onNext: A

onNext: B

...

onNext: Y

onNext: Z

onComplete

*/

Observable.just(100, 150)

    .flatMap { i ->

        Observable.interval(i.toLong(), TimeUnit.MILLISECONDS)

            .map { v -> i }

    }

    .take(10)

    .dump()

/*

onNext: 100

onNext: 150

onNext: 100

onNext: 150

onNext: 100

onNext: 100

onNext: 150

onNext: 100

onNext: 100

onNext: 150

onComplete

*/

Observable.just(100, 150)

    .concatMap { i ->

        Observable.interval(i.toLong(), TimeUnit.MILLISECONDS)

            .map { v -> i }

            .take(3)

    }

    .dump()

/*

onNext: 100

onNext: 100

onNext: 100

onNext: 150

onNext: 150

onNext: 150

onComplete

*/

Observable.interval(100, TimeUnit.MILLISECONDS)

    .switchMap { i ->

        Observable.interval(30, TimeUnit.MILLISECONDS)

            .map { l -> i }

    }

    .take(9)

    .dump()

/*

onNext: 0

onNext: 0

onNext: 0

onNext: 1

onNext: 1

onNext: 1

onNext: 2

onNext: 2

onNext: 2

onComplete

*/

• concatMap / flatMap / switchMap

• flatMapIterable

Observable.range(1, 3)

    .flatMapIterable { i -> 1..i }

    .dump()

/*

onNext: 1

onNext: 1

onNext: 2

onNext: 1

onNext: 2

onNext: 3

onComplete

*/

Observable.range(1, 3)

    .flatMapIterable<Int, Int>(

        { i -> 1..i },

        { ori, rv -> ori * rv })

    .dump()

/*

onNext: 1

onNext: 2

onNext: 4

onNext: 3

onNext: 6

onNext: 9

onComplete

*/

Observable.range(1, 3)

    .flatMapIterable<Int, Int>(

        { i -> generateSequence(1) { (it + 1).takeIf { it <= i } }.asIterable() },

        { ori, rv -> ori * rv })

    .dump()

/*

onNext: 1

onNext: 2

onNext: 4

onNext: 3

onNext: 6

onNext: 9

onComplete

*/

• RxSwift

let disposeBag = DisposeBag()

struct Player {

    var score: Variable<Int>

}

let