プロセスのclojure core.asyncパイプラインをシャットダウンする方法

Question

私はチャネルのパイプラインであるclojure処理アプリケーションを持っています。各処理ステップは、その計算を非同期的に行います（つまり、http-kitなどを使用してhttp要求を行います）。結果は出力チャネルに入れます。このようにして、次のステップはそのチャネルから読み込み、その計算を行うことができます。

私の主な機能は、この

(defn -main [args] 
(-> file/tmp-dir 
    (schedule/scheduler) 
    (search/searcher) 
    (process/resultprocessor) 
    (buy/buyer) 
    (report/reporter))) 

のように見える現在、スケジューラのステップは、（それが入力チャンネルを持っていない）のパイプラインを駆動し、ワークロードにチェーンを提供します。

私はREPLでこれを実行する場合：

(-main "some args") 

それは基本的に起因するスケジューラの無限に永遠に実行されます。 REPLからシステム全体をシャットダウンできるように、このアーキテクチャを変更する最善の方法は何ですか？各チャンネルを閉じることは、システムが終了することを意味するか？

一部のブロードキャストチャンネルは役に立ちますか？その後、ループを終了しますkill-channelに価値を置く

(def kill-channel (async/chan)) 

(defn scheduler [input output-ch kill-ch] 
    (loop [] 
    (let [[v p] (async/alts!! [kill-ch [out-ch (preprocess input)]] 
        :priority true)] 
     (if-not (= p kill-ch) 
     (recur)))) 

：

Answer 1

あなたのスケジューラのkillチャンネルとあなたのパイプラインの入力チャンネルにalts!/alts!!を持つことができます。

技術的には、プロセスを制御するためにoutput-chを使用することもできます（閉じたチャネルに返すのはfalseです）。通常、少なくともトップレベルのパイプラインの場合、明示的にキルチャネルをクリーナーにします。

（REPLで、生産の両方）を使用することが同時に、よりエレガントに、より便利にするには、Stuart Sierra's componentを使用することができ、（別のスレッドで）スケジューラ・ループを開始し、assocであなたのコンポーネントへのキルチャンネルコンポーネントのstartメソッド、次にclose!のkillチャネルを使用して、ループの終了をコンポーネントのstopメソッドで行います。

Answer 2

私は、システムのセットアップを処理するためにhttps://github.com/stuartsierra/componentのようなものを使用してお勧めします。これにより、REPLでシステムを簡単に開始および停止できるようになります。このライブラリを使用すると、各処理ステップがコンポーネントになるように設定し、各コンポーネントはstartおよびstopプロトコルのチャンネルの設定と分解を処理します。また、実装するコンポーネントごとにIStreamプロトコルを作成し、各コンポーネントをそのプロトコルを実装するコンポーネントに依存させることもできます。それはあなたに非常に簡単なモジュール性を購入します。

あなたは次のようになります。システムで終わるだろう

：

(component/system-map 
:scheduler (schedule/new-scheduler file/tmp-dir) 
:searcher (component/using (search/searcher) 
          {:in :scheduler}) 
:processor (component/using (process/resultprocessor) 
          {:in :searcher}) 
:buyer  (component/using (buy/buyer) 
          {:in :processor}) 
:report (component/using (report/reporter) 
          {:in :buyer})) 

この種のアプローチの一つの良いところは、彼らが同様のチャネルに依存している場合は、簡単にコンポーネントを追加することができたということです。たとえば、各コンポーネントがmultの内部にtapを使用してアウトチャネルを作成する場合は、プロセッサを依存コンポーネントとして扱うロギングコンポーネントだけでプロセッサ用のロガーを追加できます。

:processor (component/using (process/resultprocessor) 
          {:in :searcher}) 
:processor-logger (component/using (log/logger) 
            {:in processor}) 

私はそれがどのように動作するかのアイデアを得るためにも彼のtalkを見てお勧めします。

Answer 3

componentsというようにパイプライン要素をモデリングすることによって、Stuart Sierra's reloaded workflowを使用することを検討する必要があります。論理的なシングルトンを「クラス」としてモデル化することができます。つまり、構築と破壊（開始/停止）それぞれの1つ。