こんにちは。この記事では、ヨーロッパを駆け巡ります。つまり、リアクティブプログラミングの意味を説明し、アクター、リアクティブフローを紹介し、最後にリアクティブフローを使用して、古いOperaとその精神的な後継者であるVivaldiのように、マウスジェスチャーを認識します。

目標は、リアクティブプログラミングの基本概念を紹介し、すべてが一見すると思えるほど複雑で怖くないことを示すことです。

出所

リアクティブプログラミングとは何ですか？

この質問に答えるために、サイトに目を向けます。リアクティブアプリケーションが満たさなければならない4つの主要な基準を示す美しい画像があります。

アプリケーションは高速で、フォールトトレラントであり、拡張性に優れている必要があります。
「私たちはすべての善とすべての悪のために」のように見えますよね？

これらの言葉が意味するもの：

応答性

アプリケーションは、ユーザーに0.5秒で結果を提供する必要があります。フェイルファーストの原理もこれに起因する可能性があります。つまり、何か問題が発生した場合、「申し訳ありませんが、問題が発生しました。天気を海辺で待つよりも後で再試行してください。操作が長い場合、進行状況バーをユーザーに表示します。非常に長い場合-「あなたのリクエストは、2042年3月18日に実行されます。通知を郵送します。」
スケーラビリティは、負荷がかかったときに応答性を提供する方法です。比較的成功したサービスのライフサイクルを想像してください。
1. 起動-要求フローは小さく、サービスは1つのコアを持つ仮想マシンで実行されます。
2. リクエストのフローが増加します-カーネルが仮想マシンに追加され、リクエストは複数のスレッドで処理されます。
3. さらに負荷がかかります-バッチ処理に接続します-データベースとハードドライブへの要求はグループ化されます。
4. さらに多くの負荷-より多くのサーバーを上げて、クラスターで作業を提供する必要があります。
  理想的には、システム自体が負荷に応じて拡大または縮小する必要があります。
耐障害性

私たちは不完全な世界に住んでおり、すべてが起こることを受け入れます。システムで何か問題が発生した場合、エラー処理と回復方法を提供する必要があります
そして最後に、アーキテクチャがメッセージ駆動型メッセージングに基づいているシステムを使用して、これらすべてを達成するように招待されています。

続行する前に、イベント駆動型システムとメッセージ駆動型システムの違いについて詳しく説明します。

イベント駆動型：

イベント-システムは、特定の状態に到達したことを報告します。
イベントには多くのサブスクライバーがいる可能性があります。
通常、イベントのチェーンは短く、イベントハンドラーは（物理的にもコード的にも）ソースに近接しています。
通常、イベントソースとそのハンドラーには共通の状態があります（物理的には、情報交換に同じRAMを使用します）。

イベント駆動型とは対照的に、メッセージ駆動型システムでは：

各メッセージには1人の受信者しかいません。
メッセージは不変です。受信したメッセージの内容を変更することはできません。そのため、送信者はそれを認識して情報を読むことができます。
システムの要素は、メッセージの受信に応答する（または応答しない）ため、システムの他の要素にメッセージを送信できます。

これはすべて私たちに提供しています

俳優モデル

開発のマイルストーン：

俳優の最初の言及は1973年の科学論文-Carl Hewitt、Peter Bishop、およびRichard Steiger、「人工知能のための普遍的なモジュラーACTOR形式」
1986-アーランが登場。エリクソンは、フォールトトレランスとエラーのない伝播を提供する通信機器用の言語を必要としていました。この記事のコンテキストでは、主な機能は次のとおりです。
- すべてがプロセスです
- メッセージはコミュニケーションの唯一の方法です（Erlangは関数型言語であり、その中のメッセージは変更できません）。
..
2004-Scala言語の最初のバージョン。その特徴：
- JVMを搭載、
- 機能的
- マルチスレッドの場合、アクターモデルが選択されています。
2009-アクターの実装は別のライブラリに割り当てられました-Akka
2014-Akka.net-.Netに移植されました。

俳優は何ができますか？

アクターは同じオブジェクトですが、次のとおりです。

通常のオブジェクトとは異なり、アクターは互いのメソッドを呼び出すことはできません。
アクターは、不変メッセージを介してのみ情報を送信できます。
メッセージを受信すると、俳優は
- 新しいアクターを作成します（階層の下位になります）。
- 他のアクターにメッセージを送信し、
- 階層内および以下のアクターを停止します。

例を考えてみましょう。

アクターAは、アクターBにメッセージを送信したいと考えています。彼が持っているのは、ActorRef（アドレス）だけです。アクターBはどこでもかまいません。
アクターAは、システム（ActorSystem）を介して文字Bを送信します。システムは、アクターBのメールボックスにレターを入れ、アクターBを「ウェイクアップ」します。アクターBは、メールボックスからレターを受け取り、何かを実行します。

別のオブジェクトでメソッドを呼び出すことに比べて、それは不必要に複雑に見えますが、アクターが特定の刺激に反応して何かをするように訓練された人だと想像すると、アクターのモデルは現実の世界に完全に適合します。

父と息子を想像してください：

父親は息子のSMSkuに「部屋の掃除」を送り、自分のことを続けます。息子はSMSkuを読み、クリーニングを開始します。一方、父はポーカーをしています。息子は掃除を終了し、SMS「完了」を送信します。簡単そうですね。

ここで、父と息子は俳優ではなく、互いのメソッドを引っ張る普通のオブジェクトであると想像してください。父親は「部屋を掃除する」方法のために息子を引っ張り、息子が掃除を終えて父親にコントロールを戻すまで待つかかとを追います。現時点では、父親はポーカーをプレイできません。このコンテキストでは、アクターモデルがより魅力的になりつつあります。

では、に移りましょう

Akka.NET

以下に書かれていることはすべて、JVMの元のAkkaに当てはまりますが、私にとっては、C＃はJavaよりも近いため、Akka.NETを例として使用します。

Akkaの利点は何ですか？

メッセージングを介したマルチスレッド。あらゆる種類のロック、セマフォ、ミューテックス、および共有メモリを使用する従来のマルチスレッドに特有のその他の魅力に悩まされる必要はもうありません。
システムとそのコンポーネント間の透過的な通信。複雑なネットワークコードを心配する必要はありません-システム自体がメッセージの宛先を見つけて、メッセージの配信を保証します（ここでUDPとTCPについてのジョークを挿入できます）。
自動的にスケールアップまたはスケールダウンできる柔軟なアーキテクチャ。たとえば、負荷がかかると、システムは追加のクラスターノードを発生させ、負荷を均等に分散できます。

しかし、スケーリングのトピックは非常に広範囲であり、別の出版物に値します。したがって、すべてのプロジェクトで役立つ機能についてのみ詳しく説明します。

エラー処理

アクターには階層があり、ツリーとして表すことができます。各俳優は親を持ち、「子供」を持つことができます。

Akka.NETドキュメント Copyright 2013-2018 Akka.NETプロジェクト

各アクターに対して、監督戦略を設定できます。「子供」に何か問題が発生した場合の対処方法です。たとえば、問題のある俳優を「打ち負かし」、同じタイプの新しい俳優を作成し、同じ仕事を彼に任せます。

たとえば、Akka.net CRUDでアプリケーションを作成しました。このアプリケーションでは、「ビジネスロジック」の層がアクターに実装されています。このプロジェクトの目的は、スケーラブルでないシステムでアクターを使用する必要があるかどうか、つまり、役者が生活を改善するのか、痛みを増すのかを調べることでした。

Akkaの組み込みエラー処理がどのように役立つか：

GIF

すべてが正常で、アプリケーションが機能し、
リポジトリに何かが起こりましたが、5回のうち1回だけ結果が出ます
私は、監督戦略を「毎秒10回試行」に設定しました。
アプリケーションは（ゆっくりではあるが）再び動作し、問題が何であるかを理解する時間があります。

「さあ、このようなエラー処理を自分で書いてみよう。なぜ誰かがミスをするのか？」公正な発言ですが、障害点が少ない場合のみです。

そして、少しのコード。これが、IoCコンテナー内のアクターシステムの初期化の様子です。

public Container() { system = ActorSystem.Create("MySystem"); var echo = system.ActorOf<EchoActor>("Echo"); //stop initialization if something is wrong with actor system var alive = echo.Ask<bool>(true, TimeSpan.FromMilliseconds(100)).Result; container = new WindsorContainer(); //search for dependencies //register controllers //register ActorSystem propsResolver = new WindsorDependencyResolver(container, (ActorSystem)system); system.AddDependencyResolver(propsResolver); actorSystemWrapper = new ActorSystemWrapper(system, propsResolver); container.Register(Component.For<IActorRefFactory>().Instance(actorSystemWrapper)); container.Register(Component.For<IDependencyResolver>().Instance(propsResolver)); }

EchoActorは、送信者に値を返す最も単純なアクターです。

  public class EchoActor : ReceiveActor { public EchoActor() { Receive<bool>(flag => { Sender.Tell(flag); }); } }

アクターを「通常の」コードに接続するには、Askコマンドを使用します。

  public async Task<ActionResult> Index() { ViewBag.Type = typeof(Model); var res = await CrudActorRef.Ask<IEnumerable<Model>>(DataMessage.GetAll<Model>(), maxDelay); return View(res); }

合計

俳優とスニッキー、私は言うことができます：

スケーラビリティが必要な場合は、それらを見てください。
複雑なビジネスロジックの場合は、次の理由で使用しない方が良い
- 奇妙な依存性注入。必要な依存関係でアクターを初期化するには、まずPropsオブジェクトを作成し、それをActorSystemに渡して目的のタイプのアクターを作成する必要があります。 IoCコンテナー（Castle WindsorやAutofacなど）を使用して小道具を作成するには、既製のラッパー-DependencyResolversがあります。しかし、IoCコンテナーが依存関係の有効期間を制御しようとしていて、しばらくしてシステムが静かに落ちたという事実に直面しました。
  
  *おそらく、オブジェクトに依存関係を注入する代わりに、この依存関係を子アクターとして配置する必要があります。
- タイピングの問題。 ActorRefは、それが参照するアクターのタイプについて何も知りません。つまり、コンパイル時には、アクターがこのタイプのメッセージを処理できるかどうかはわかりません。

パート2：ジェットストリーム

次に、より一般的で便利なトピックであるジェットフローに移りましょう。プロセスでアクターと会うことができない場合、フロントエンドとバックエンドの両方でRxストリームが確実に役立ちます。それらの実装は、ほとんどすべての最新のプログラミング言語で行われています。今日ではバックエンドのプログラマでさえJavaScriptで何かをしなければならないことがあるため、RxJの例を挙げます。

Rxストリームは、すべての一般的なプログラミング言語で使用できます。

CC BY-NC 4.0でライセンスされているAndre Staltzの「不足しているリアクティブプログラミングの概要」

ジェットストリームとは何かを説明するために、プルコレクションとプッシュコレクションから始めます。

	単一の戻り値	複数の戻り値
引く同期インタラクティブ	T	IEnumerable <T>
プッシュ非同期リアクティブ	タスク<T>	IObservable <T>

プルコレクションは、プログラミングで私たちが慣れているものです。最も顕著な例は配列です。

 const arr = [1,2,3,4,5];

すでにデータがあり、彼自身はこのデータを変更しませんが、リクエストに応じて提供できます。

 arr.forEach(console.log);

また、データを処理する前に、何らかの方法でデータを処理できます。

 arr.map(i => i+1).map(I => “my number is ”+i).forEach(console.log);

ここで、最初はコレクションにデータがないことを想像してみましょうが、データが表示されたことは間違いなく通知されます（プッシュ）。同時に、必要な変換をこのコレクションに適用できます。

例：

 source.map(i => i+1).map(I => “my number is ”+i).forEach(console.log);

ソースに1などの値が表示されると、console.logは「my number is 1」を出力します。

仕組み：

新しいエンティティが表示されます-件名（または観測可能）：

 const observable = Rx.Observable.create(function (observer) { observer.next(1); observer.next(2); observer.next(3); setTimeout(() => { observer.next(4); observer.complete(); }, 1000); });

これは、状態の変化に関する通知を送信するプッシュコレクションです。

この場合、4秒後に1、2、3の数字がすぐに表示され、コレクションは「完了」します。これはこのような特別なタイプのイベントです。

2番目のエンティティはObserverです。彼はサブジェクトイベントにサブスクライブし、受信したデータで何かをすることができます。例：

 observable.subscribe(x => console.log(x)); observable.subscribe({ next: x => console.log('got value ' + x), error: err => console.error('something wrong occurred: ' + err), complete: () => console.log('done'), }); observable .map(x => 'This is ' + x) .subscribe(x => console.log(x));

1つのサブジェクトが多くのサブスクライバーを持つことができることがわかります。

簡単に見えますが、なぜこれが必要なのかはまだ明確ではありません。リアクティブフローを使用する際に知っておく必要のある定義をさらに2つ示し、その後、実際にそれらがどのように機能し、どのような状況でその潜在能力が最大限に発揮されるかを示します。

冷たい観測可能量

誰かがサブスクライブしたときにイベントについて通知します。
サブスクリプションの時間に関係なく、データストリーム全体が各サブスクライバーに再度送信されます。
データは、サブスクライバーごとにコピーされます。

これはどういう意味ですか：会社（被験者）がギフトの配布を手配することにしたとしましょう。各従業員（オブザーバー）は仕事に来て、ギフトのコピーを受け取ります。誰も取り残されていません。

ホットオブザーバブル

サブスクライバの存在に関係なく、イベントについて通知しようとします。イベントの時点でサブスクライバーがいなかった場合、データは失われます。

例：朝、従業員向けのホットケーキが会社に持ち込まれます。彼らが連れて来られると、すべてのひばりは匂いで飛んで、朝食のためにパイを作ります。しかし、後に来たフクロウはもはやパイを取得しません。

どのような状況でジェットストリームを使用しますか？

時間の経過とともに配信されるデータストリームがある場合。たとえば、ユーザー入力。または、任意のサービスからのログ。プロジェクトの1つで、N秒でイベントを収集し、パック全体を同時に記録する自己記述型のロガーを見ました。バッテリーコードがページを占有しました。 Rxストリームが使用された場合、はるかに簡単になります。

RxJs Reference / Observable 、CC BY 4.0の下でライセンスされたドキュメント 。
（リアクティブフローを使用したさまざまな操作の実行内容を説明する多くの例と写真があります）

 source.bufferTime(2000).subsribe(doThings);

そして最後に、使用例。

Rxストリームでのマウスジェスチャーの認識

古いOperaまたはその精神的な後継者であるVivaldiには、マウスジェスチャーを使用したブラウザコントロールがありました。

Gif-Vivaldiでのマウスジェスチャー

つまり、マウスの上下、左右の動き、およびそれらの組み合わせを認識する必要があります。これはRxストリームなしで記述できますが、コードは複雑で維持が困難です。

そして、Rxストリームでの表示は次のとおりです。

最後から始めます-元のシーケンスで検索するデータと形式を設定します。

 //gestures to look for const gestures = Rx.Observable.from([ { name: "Left", sequence: Rx.Observable.from([{ x: -1, y: 0 }]) }, { name: "Right", sequence: Rx.Observable.from([{ x: 1, y: 0 }]) }, { name: "Up", sequence: Rx.Observable.from([{ x: 0, y: -1 }]) }, { name: "Down", sequence: Rx.Observable.from([{ x: 0, y: 1 }]) }, { name: "Down+Up", sequence: Rx.Observable.from([{ x: 0, y: 1 }, { x: 0, y: -1 }]) }, { name: "Up+Right", sequence: Rx.Observable.from([{ x: 0, y: -1 }, { x: 1, y: 0 }]) } ]);

これらは単位ベクトルとその組み合わせです。

次に、マウスイベントをRxストリームに変換する必要があります。すべてのRxライブラリには、標準イベントをObservableに変換するための組み込みツールがあります。

 const mouseMoves = Rx.Observable.fromEvent(canvas, 'mousemove'), mouseDowns = Rx.Observable.fromEvent(canvas, 'mousedown'), mouseUps = Rx.Observable.fromEvent(canvas, 'mouseup');

次に、マウスの座標を2でグループ化し、それらの差を見つけて、マウスオフセットを取得します。

 const mouseDiffs = mouseMoves .map(getOffset) .pairwise() .map(pair => { return { x: pair[1].x-pair[0].x, y: pair[1].y-pair[0].y } });

イベント「mousedown」と「mouseup」を使用してこれらの動きをグループ化します。

 const mouseGestures = mouseDiffs .bufferToggle(mouseDowns, x => mouseUps) .map(concat);

concat関数は、短すぎる動きを切り取り、方向がおおよそ揃っている動きをグループ化します。

 function concat(values) {//summarize move in same direction return values.reduce((a, v) => { if (!a.length) { a.push(v); } else { const last = a[a.length - 1]; const lastAngle = Math.atan2(last.x, last.y); const angle = Math.atan2(vx, vy); const angleDiff = normalizeAngle(angle - lastAngle); const dist = Math.hypot(vx, vy); if (dist < 1) return a;//move is too short – ignore //moving in same direction => adding vectors if (Math.abs(angleDiff) <= maxAngleDiff) { last.x += vx; last.y += vy; } else { a.push(v); } } return a; }, []); }

X軸またはY軸に沿った動きが短すぎる場合、ゼロにリセットされます。そして、得られた変位座標から符号のみが残ります。したがって、探していた単位ベクトルが取得されます。

 const normalizedMouseGestures = mouseGestures.map(arr => arr.map(v => { const dist = Math.hypot(vx, vy);//length of vector vx = Math.abs(vx) > minMove && Math.abs(vx) * treshold > dist ? vx : 0; vy = Math.abs(vy) > minMove && Math.abs(vy) * treshold > dist ? vy : 0; return v; }) ).map(arr => arr .map(v => { return { x: Math.sign(vx), y: Math.sign(vy) }; }) .filter(v => Math.hypot(vx, vy) > 0) );

結果：

 gestures.map(gesture => normalizedMouseGestures.mergeMap( moves => Rx.Observable.from(moves) .sequenceEqual(gesture.sequence, comparer) ).filter(x => x).mapTo(gesture.name) ).mergeAll().subscribe(gestureName => actions[gestureName]());

sequenceEqualを使用すると、受信した動きを元の動きと比較し、一致する場合は特定のアクションを実行できます。