はじめに

本記事はサムザップAdvent Calendar 2017の1３日目の記事です。
昨日は@takahashi_wataruさんのUnity2017.2.0p3で.Net4.6を使うとVisualStudioのインテリセンスが効かない対処法+αでした！

今年の9月からサムザップでお世話になっている@mmm_hiroです。

現在関わっているプロジェクトのいろいろによりPython3（使用歴半年）を利用しているのですが
今回はPythonでRxPYを利用したリアクティブプログラミングについて記事を書こうかなと思います。

他の言語に比べて、PythonにおけるReactiveXの記事は少なく、自分もコツを掴むまでに多少の時間を必要としました。
まずはネット上にある有用な記事（言語問わず）を読み、リアクティブプログラミングの概念を把握するところから初めましょう

http://reactivex.io/
- 様々な言語でリアクティブプログラミングを実装
https://github.com/ReactiveX/RxPY
- その中のPython実装がこのライブラリ

ReactiveXを理解するために何を読めばよいか？

ネットにあげて下さった記事をひたすら熟読していきましょう。全部読めばなんとなく理解できてくるはずです。

https://github.com/ReactiveX/RxPY/blob/develop/notebooks/Getting%20Started.ipynb
- notebookで解説されているので丁寧に見ていけば概要を把握できると思います。
http://ninjinkun.hatenablog.com/entry/introrxja
- Python実装ではないでうが、Twitterの「フォローするアカウントを推薦するUIエレメント」を例にした実装手順はとても参考になります。
https://qiita.com/acple@github/items/8d3a4d3414fa59adff70
- cold hotについて
https://qiita.com/acple@github/items/6cfee916f09632037a6e
- こわくないReactive Extensions超入門

導入

pip install rx

サンプルをかけるだけ書く

以下、当日、かけるところまでサンプルを書き続けてみる。以下当日中随時更新していきます（2017/12/12、21:05記述中）

一番シンプルなもの

# 値を一個ずつ送信、2倍してプリント
xs = Observable.from_([1,2,3])
d = xs.map(lambda x:x*2). \
    subscribe(print)

2
4
6

ちなみに、配列内のデータをそのまま送ってるわけではないことに注意。

xs = Observable.from_([[1,2,3],[4,5,6]])
d = xs.map(lambda x:x*2). \
    subscribe(print)

こちらの場合は[1,2,3]と[4,5,6]が一つのストリームとして送信されています。

[1, 2, 3, 1, 2, 3]
[4, 5, 6, 4, 5, 6]

指定の条件に満たない場合は除去する。

from rx.subjects import Subject
print("50以下は無効化してしまう壁\n")

stream = Subject()
d = stream.do_action(lambda x: print(x)) \
    .filter(lambda param: param>50) \
    .subscribe(lambda result: print("壁を貫通！{0}のダメージ".format(result)))

# 攻撃！
stream.on_next(0)
stream.on_next(50)
stream.on_next(51)
stream.on_next(49)
stream.on_next(100)
stream.on_next(-50)
stream.dispose()

50以下は無効化してしまう壁

0
50
51
壁を貫通！51のダメージ
49
100
壁を貫通！100のダメージ
-50

stream_left,stream_right両方に値が来たときだけsubscribeが動作

from rx.subjects import Subject

stream_right = Subject()
stream_left = Subject()

stream = stream_left.zip(stream_right, lambda x, y: x+y)
d = stream.subscribe(lambda result: print("合算：{}".format(result)))

stream_left.on_next(1)
stream_left.on_next(2)
stream_right.on_next(3)
stream_right.on_next(4)

合算：4
合算：6

指定した回数しかsubscribeに到達しないようにする

from rx.subjects import Subject

stream = Subject()
stream.take(3).subscribe(lambda member: print("{}".format(member)))

# 
print("先着順です！商品獲得は以下の3名です")
stream.on_next("1さん")
stream.on_next("2さん")
stream.on_next("3さん")
stream.on_next("4さま")
stream.on_next("5さま")

先着順です！商品獲得は以下の3名です
1さん
2さん
3さん

明日は@norimatsu_yusukeさんのWPFのビューについてです。

RxPYをひたすら素振りする