Scala + Akka + SprayでIMAPクライアントを開発するプロセスにおける「暗号化されたパイプライン」の複雑さ

最近では、愛用のオブジェクト指向C ++から新しいものに切り替えましたが、機能的なScalaはまだ明確ではありません。 移行の理由は、まったく異なる話です。 しかし、それらの1つは、Akkaライブラリを使用して、レビュー、俳優モデルのサポートから判断して、十分な存在感でした。 私は長い間、この技術のすべての記述された利点を自分の経験で試してみるのを夢見てきました。 俳優モデルの最も標準的な（私の意見では）決定-Erlang-私はそれをマスターするのに時間がかかりすぎると思ったので、私は却下しました。言語。 したがって、結果として、特に私がかつて長期間Scalaを勉強し始めたが、不便さのためにそれを放棄したので、私の選択はAkkaとともにScalaに落ちました。 しかし、結局のところ、今回は実験に最適な時期を選択しませんでした。これは、プロジェクトのかなり堅実な部分が既に完了した後に初めて確信しました。

開始する

開発は最初から加速されており、私のアプリケーションに必要なほとんどすべての必要な機能は、サードパーティのライブラリとしてインターネット上で豊富に利用できました。 また、Scala専用にまだ記述されていないものの不足は、Javaのさまざまなコンポーネントの豊富な多様性によって補われています。 ただし、問題は、いつものように、予期しない場所で発生しました。



実際、ある時点で、アプリケーションはIMAPサーバーに接続して、受信したメールメッセージを読み取って処理する必要があります。 そして、アクターモデルはネットワークとの非同期作業を意味するため、新しいアプリケーションの構造に美しく有機的に適合するために、メールサーバーに接続してメールを非同期に受信できるライブラリが必要でした。 短い検索の後、akka-camelモジュールに出会いました。このモジュールを使用すると、俳優のメッセージチャネルとしてapache-camelライブラリを使用できます。 そして、ラクダは、判明したように、とりわけメールサーバーに接続できます。 さらに、必要な接続パラメーターを指定する場合、ラクダは新鮮な（\ Recent）フラグの読み取り、読み取りメッセージの削除、または特別に作成されたフォルダーへのコピー/移動のみを行うことができます。 もっと夢を見ることさえできませんでした。 SBTを開始するには、依存関係akka-camel、camel-core、およびcamel-mailについて言及するだけでした。

最初の試み

アクターを作成し、それをIMAPサーバーに接続するには、ほんの数行のコードが必要でした。 そして、アプリケーションログでは、メッセージのテキストが落ちました。私自身がテストのためにメールに送信しました。 私はすでに満足して手をこすり始め、次のタスクについて考え始めましたが、結果として処理のためにメールを受け取るワークボックスに接続しようとする場合のためだけに決めました。 そして、ここで私の俳優は例外を投げて「倒れた」。 結局のところ、彼はサーバーの応答を正しく解析できませんでした。 インターネットでは、このエラーと可能な解決策に関する情報は見つかりませんでした。 そして少し落ち込んだ。 私はプロトコルの仕様を勉強したり、クライアントを書いたりすることに時間を費やしたくありませんでした。 そして、しぶしぶ、時間を節約するという名目で、私は完全な非同期の意図されたコースから戻って、同期JavaMailブロッキングライブラリを使用することにしました。 ただし、同じ場所で、同じ例外を除いて、このライブラリは落ちました。 その後、理想を捨てることが弱虫や怠zyな人々の道であると固く決心しましたが、私は非同期性とアクターを使って自分のIMAPクライアントを作成しています。 さらに、IMAP全体を実装する必要はありませんでした。機能は非常に制限されていました。承認、INBOXフォルダーの選択、メッセージのリストの受信、特定のメッセージの読み取り、メッセージの別のフォルダーへのコピー、削除。

二度目の試み

私は特にどこから始めるかを選択する必要はありませんでした。 ご存じのように、Akka開発者はある時点で、ネットワークI / OのNettyを放棄し、Sprayを支持しました。 将来、AkkaとSprayの開発は非常に密接に絡み合っていたため、それらのドキュメントでさえ相互に参照され、spray.ioのコードはakka.ioにスムーズに移行しました。 そして、ここで主な問題は私を待っていました：かつて、バージョン2.xの開発中に、AkkaはSprayで使用されるチャネルのアイデアを採用しました（これらは「パイプライン」でもあり、英語のパイプラインです）。著者によれば、背圧をサポートするネットワークプロトコルを作成するのは簡単です。つまり、受信者が送信者からのデータストリームを処理、フィルタリング、分離、乗算する時間がない場合にパイプが詰まらないように「バルブを締める」機能です。それらでやらないでください。 しかし、これらの「パイプライン」の何かがおかしくなり、実験段階を離れることなく、廃止されると宣言されました。 Akkaから最後に発表されたイノベーションは、チャネルを完全に置き換えることを目的としていますが、「リアクティブストリーム」は既にハブで記述されています しかし、この革新はまだ発表段階にあるため、akka 2.3.6の最新バージョンではまだありませんが、もうチャネルはありません。 チャンネルはスプレー内に残りましたが、それらに関するすべてのドキュメントは古いAkkaバージョン2.2.0-RC1ドキュメントにつながり、現在のすべての現実を反映しなくなりました。 新しいAkkaのドキュメントでは、チャンネルはスプレーにとどまっていると書かれています。 一般に、私のメールクライアントの最初のバージョンは、フランケンシュタインの辛抱強い子供とほぼ同じであることが判明しました。 私はこの概念が非常に複雑であるように思えたため、すぐに「パイプライン」を放棄することにしました。したがって、クライアントはサーバーからの文字列をByteStringの形式で直接操作しました。 より正確には、このストリームのフラグメントを使用すると、関心のある回答が1つにまとまるか、2つの回答が一致しないことを誰も保証しないためです。 奇跡的な方法で、多数のマティオットがモニターに注がれ、コードの一部が書き直されたため、さまざまな場所で見つかったいくつかのコードを使用して、アクターにSSL / TLS暗号化をねじ込むことができました。 公式ドキュメントの多くのバージョンの一部（非常に古い）の特定のコードのみで見つかったコードは、機能しませんでした。



その控えめな機能の後続の各段階の実装により、私のクライアントはより怪しくなりました。 結局、次の繰り返しである午前3時に、私はあなたがそのように生きることができなくなったことに気づきました。同じブロッキングJavaMailを試すために迷惑をかけてTODOに書きました。

3回目の試み

しかし、翌朝（または昼食時に）頑固なスカムバッグであったため、JavaMailを使いこなそうとする代わりに、最初にgithubのSprayソースにアクセスしました。 数日かけてそれらを研究し、受け取った情報を自分のニーズに合わせて調整しましたが、費やした時間は見事に報われました。 まず、ソースコードで、ドキュメントの他の部分で説明されていないConnectionHandlerクラスに出会いました。 スプレー缶でこのクラスのアプリケーションを研究しているときに、これらの「パイプライン」を使用する方法と場所を理解しました。ドキュメントから、解決するために呼び出されるタスクだけを見つけましたが、 それを行う方法はわかりませんでした。 同じソースコードで、パイプを「接続」する方法、つまり、パイプライン（PipelineStage）の複数の「パイプ」（ステージ）を1つの共通の「パイプライン」に結合する方法、これがもたらすもの、使用方法を発見しました。 また、私は前日にSSL暗号化が失敗した理由と方法がどのように機能するかを見つけました。その瞬間までは、「うまく機能し、登る必要はありません」というブラックボックスのままでした。

啓発

詳細に興味がある人のために：「パイプライン」はパーツで構成され、オリジナルでは「ステージ」または「ステージ」（英語のステージ）と呼ばれますが、画像を維持するために「パイプ」と呼びます。 コード内のこれらの「パイプ」は>>演算子を使用して結合され、順序が重要です。 最初はクライアントに最も近い「パイプ」で、最後はサーバーに近いものです。 つまり、クライアントから送信されるものはすべて、「パイプライン」を左から右へ、サーバーから-逆に右から左へと通過します。 たとえば、暗号化を実行する「パイプ」は最後に示されます。したがって、クライアントが「パイプ」に送信するものはすべて必要な変換をすべて経てから暗号化され、暗号化されたデータが最終的にサーバーに送信されます。 またその逆に、サーバーが送信するすべてのものが最初に復号化され、次に「パイプライン」の残りによって変換されます。 なぜこの配管がすべて必要なのですか？ さまざまなものに。 たとえば、送信または受信されたデータをフィルタリングします。 または、一部のエンティティを他のエンティティに変換します。これは、プロトコルを実装するときに役立ちます。 特定のケースクラスDeleteMessage（id：String）があるとします。 クライアントはDeleteMessage（「23」）のインスタンスを「パイプライン」に送信し、1つの段階（「パイプ」の1つ）でこのクラスはサーバーが理解するコマンド「a001 STORE 23 + FLAGS.SILENT（\ Deleted）」に変換されます。 それでも、たとえば、サーバーからの応答が不完全で、追加が予想される場合、「パイプ」はデータの配信を遅延させる可能性があります。



最初に私を完全に混乱させた主な点は、「イベント」と「コマンド」の概念概念の存在と、それに対応するパイプラインの内訳、イベントパイプラインとコマンドパイプライン1つのクラス：PipelineStage。 コンセプトの最初の段階で私に理解されなかったのはこれらの人々でした（まあ、マニュアルは散らばっていて分かりにくい、敗者だけが最後まで読んで、普通の人はすぐに行き、コーンを埋めます）私はパイプラインを悪いと思い、それが多すぎると決めました複雑で時間を費やす価値はありません。 これは「バックプレッシャー」と何らかの関係があるように思われました。これは考慮に入れて実装する必要がありますが、まったく必要ありませんでした。 そして、これは、最初に1つの「パイプ」を貼り付ける場所、またはサーバーに何かが届くように何かを入れる方法がわからなかったという事実に加えて。 そして、彼女から答えを得る方法。 そして、これらのパイプがさらに2つありました。 一方、もしこの誤解がなければ、私の小さな怪物が発明された後の「配管」アプローチの力を十分に感じたことはなかっただろう。 実際、アイデアは非常にシンプルであることがわかり、私を笑わせることさえできました。イベントはサーバーからクライアントに送られるものであり、コマンドはクライアントからサーバーに送られるものです。 その結果、パイプは1つであることが判明しました。パイプの内部では、パイプ自体が2つの近づいてくるフローを分離しているため、どこから来るのか混乱しないようになっています。

結果

一般に、私の研究の結果、IMAPサーバーへの接続を担当する新しいクラスを作成しました。

class Connection(client: ActorRef, remoteAddress: InetSocketAddress, sslEncryption: Boolean, connectTimeout: Duration)(implicit sslEngineProvider: ClientSSLEngineProvider) extends ConnectionHandler { actor => override def supervisorStrategy = SupervisorStrategy.stoppingStrategy def tcp = IO(Tcp)(context.system) log.debug("Attempting connection to {}", remoteAddress) tcp ! Tcp.Connect(remoteAddress)//, timeout = Some(Duration(connectTimeout, TimeUnit.SECONDS))) context.setReceiveTimeout(connectTimeout) val pipeline = eventFrontend >> ResponseParsing() >> SslTlsSupport(512, publishSslSessionInfo = false) override def receive: Receive = { case connected: Tcp.Connected => val connection = sender() connection ! Tcp.Register(self, keepOpenOnPeerClosed = sslEncryption) client ! connected context.watch(connection) context.become(running(connection, pipeline, pipelineContext(connected))) case Tcp.CommandFailed(_: Tcp.Connect) => throw new ConnectionFailure(1, "Failed to connect to IMAP server") case ReceiveTimeout => log.warning("Connect timed out after {}", connectTimeout) throw new ConnectionFailure(2, "Connect timed out") } def eventFrontend = new PipelineStage { def apply(context: PipelineContext, commandPL: CPL, eventPL: EPL): Pipelines = new Pipelines { val commandPipeline: CPL = commandPL val eventPipeline: EPL = { case event => client ! event } } } def pipelineContext(connected: Tcp.Connected) = new SslTlsContext { def actorContext = context def remoteAddress = connected.remoteAddress def localAddress = connected.localAddress def log = actor.log def sslEngine = if (sslEncryption) sslEngineProvider(this) else None } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このクラスは、spray.can.client.HttpClientConnectionクラスを単純化することで取得しました。 これは、spray.io.ConnectionHandlerから継承され、それはakka.Actorから継承されます。 つまり、彼は普通の俳優です。 実際、これは配管工です。彼をスタニスラフと呼びましょう。 スタニスラフは、クライアントからサーバーへのパイプラインと、このパイプラインを介したデータのやり取りを担当しています。 標準のcontext.actorOf（...）でアクター（つまり自分）を初期化するときに、このパイプラインを舗装します。 パイプラインプロパティは、この場合、3つのパイプ（フロントエンド、サーバーからの応答パーサー、およびSSL / TLSエンクリプター）から組み立てられたパイプラインです。 そして今、配管工スタニスラフに送信されたすべてのデータ（彼を、俳優として、オペレーターにメッセージを送信することによって！、tellメソッド、または他の利用可能な手段によって）、彼はそれを慎重にパイプに入れてサーバーに送信します。 また、サーバーからの応答に含まれるすべてのものもパイプから慎重に取り出され、クライアントに送信されます。 彼は私たちの勤勉な男スタシクです。



私が使用したパイプについては。



SslTlsSupportはSprayの標準SSL / TLS暗号化機能です。 特別なコンテキスト（pipelineContextメソッドによって返される）が必要です。また、サーバー側で接続を閉じた後でも、接続のクライアント側を開いたままにする必要があります（いわゆるセミオープン接続）。



ResponseParsingは、構文解析を担当する「パイプ」のインスタンスを返すapply（）関数を使用して既に書き込まれたオブジェクトです。サーバーからの文字ストリームを（「生の」Tcp.Receivedメッセージの形で）特定の回答のケースクラスに解析し、処理しますすでにターゲットアクター（私のIMAPクライアント）。 パーサーは、返されたデータの整合性を監視する役割も果たします。サーバーからの応答が完了していない場合は追加データを待機し、それらが1つになった場合は複数の応答を互いに分離します。 これにより、クライアントのコードが大幅にアンロードされました。クライアントは、ひどいパッチワークモンスターから、シンプルでわかりやすく、直接的で複雑でない男Vasilyに変わりました。 Vasinaのパフォーマンスを維持するために必要なテストの量も大幅に削減されました。



最後に、 eventFrontendは、「パイプ」のインスタンスを返す関数です-PipelineStage 、その本質は1つです。すべての「イベント」（つまり、サーバーからパイプライン全体を通過し、すでに必要なすべての変更が行われたデータ）をクライアント、つまりVasyaに送信します。クラスコンストラクターに渡された変数のおかげで、スタニスラフが知っているアドレス。



このような必要性がないため、コマンドの特別なレンダリングは行いませんでした。 サーバーへのすべてのコマンドは、単純なTcp.Writeを使用して送信されます。

エピローグ

ここで、実際には、すべての配管。 エピローグとして、クライアント自体はAkka.FSMに基づく有限状態マシンであると言えます。 彼のためにオートマトンとユニットテストを書くのはとてもエキサイティングなミニゲームであるため、私はAkkaでのこのコンセプトの実装に夢中になりました。