問題タブ [enterprise-integration]

Spring Integration を使用して、ディレクトリから入力メッセージを取得する必要があります。これは、読み取り専用アクセスしかできない CIFS / Samba 共有フォルダーであるため、削除する必要がなく、新しいファイルのみを取得する方法を見つける必要があります。デフォルトの AcceptOnceFileListFilter の動作は、コンテナを再起動しない場合にのみ機能するため、理想的ではありません。

私が考えているのは、ファイル ポーラーが X 分前に作成されたファイルのみを取得する方法です。私はまだ繰り返しファイルを拾っていますが、はるかに短い期間からです (私はそれに対処できます)。

私がする必要があるのは、独自の FileListFilter を実装して、X 分前に作成されたファイルのみを返すことだと思います。デフォルトのSpringフィルター内または他の場所に、その実装が既にあるかどうかを尋ねたいだけです。より良い提案も大歓迎です！

ありがとうウォン

一般的なアプローチは、メッセージを発行する前にコンシューマ キューを作成することですが、ここでは非常に緊密な結合が行われているようです。一般に、この交換から消費されるキューの数と、それぞれが取得するメッセージの量を知りたくありません。サーバーを消費者から切り離しながら信頼性を維持しますか?

このイメージでは、認証リクエストを処理するために使用できるサーバー キューがあるかどうかをクライアント アプリが認識する必要があります。キューがない場合、メッセージはデッドレターになるため、これについて誰も考えたことはないと思いますか? それとも、このキューが常に存在すると仮定します (永続的です)?

アウトバウンドアダプターの直前にアグリゲーターがあります (リクエストをバッチ処理するため)。

アウトバウンド アダプターに障害が発生した場合、アグリゲーターがメッセージのリリースを停止し、最終的にアウトバウンド アダプターがオンラインに戻るまで、背圧が統合フローに伝播するようにしたいと考えています。

これは、EIP 内で望ましいパターンであるように思われます。これには、キューに基づくチャネルが含まれていることは確かですが、いくつかの理由から、これがどのように実装されているかはわかりません。

A) アダプターの失敗の開始が例外によって暗示されている場合 (実際@retryable、送信アダプターで使用しています)、メッセージを再キューイングする「正しい」方法は何ですか?

B)アグリゲータのリーパー スレッドがダウンストリーム キューの「背圧」を尊重すると仮定しています。とはいえ、そのような状況でアグリゲーターにアップストリームからのメッセージの消費を停止するように指示するにはどうすればよいでしょうか?

CamelConfiguration15 を構成するがありますRoutes。

私が達成したいMessageのは、ルート内のすべての着信( )のヘッダーをチェックし、exchange.getHeaders()存在しない場合はヘッダーを追加することです。

Processoreach 内で使用してそれを達成できますRouteBuilder。元。

目的は、すべての交換メッセージ間で同じ ID を使用することです。これにより、後でそれらを簡単に関連付けることができます。

それで、すべてのsCamelConfiguration#configureに適用されるように、これを行う Camel の方法はありますか。Route

以下のようにインターセプトを期待していました。

.to()傍受されますが、それぞれで続行されないようですRouteBuilder。

参考文献

http://camel.apache.org/intercept.html

http://www.davsclaus.com/2009/05/on-road-to-camel-20-interceptors-round.html

MongoDb からオブジェクトを取得して現在のペイロードに追加し、別のデータベースに保存しようとしています。

元のオブジェクトを保持するIn.bodyからクエリをエンリッチする問題...

では、クエリ ( {"entity.id": ""} ) をenrich(mongoldb:...)に渡し、元のオブジェクトを保持して結果とマージするにはどうすればよいですか?

ありがとう。

RSS/Atom フィード アグリゲーターを実装しようとしてspring-integrationいますが、主に Java DSL を使用してIntegrationFlow. このアグリゲーターの要件は、実行時にフィードを追加/削除できることです。つまり、フィードは設計時にはわかりません。

Feed.inboundAdapter()基本をテスト URL で使用し、トランスフォーマーを使用してフィードからリンクを抽出し、それを に渡してoutbound-file-adapterリンクをファイルに保存するのは簡単であることがわかりました。inbound-file-adapterただし、実行中のファイルから (数千の) フィード URL を読み取り、フィード URL を含むFileSplitter各結果を渡しMessage<String>て新しい を登録しようとすると、非常に行き詰まりましたFeed.inboundAdapter()。これは Java DSL では不可能ですか?

理想的には、次のことができれば大好きです。

春の統合Java DSLコードを何度も読んだ後（そして途中でたくさんのことを学んだ後）、この方法でそれが可能であることがわかりません。それで… A)そうですか？B) そうあるべきか? C) 提案は?

.handle(new FileSplitter())の出力を取得して渡すことができるはずのように感じます.handleWithAdapter(Feed.inboundAdapter(/*stuff here*/))が、DSL は s しか参照していませんoutbound-adapter。インバウンドアダプターは実際には単なるサブクラスでAbstractMessageSourceあり、それらのいずれかを指定できる唯一IntegrationFlows.from(/*stuff here*/)の場所はメソッドの引数としてです。

ファイルから入力を取得し、行ごとに分割し、その出力を使用してインバウンド フィード アダプターを登録し、それらのフィードをポーリングし、表示されるフィードから新しいリンクを抽出してファイルに追加することが可能であると考えていました。 . そうではないかのように見えます。

これを機能させるためにできる賢いサブクラス化はありますか??

それが失敗すると...そしてこれが答えになると思います.春の統合動的Ftpチャネルリゾルバーの例と、それを適応させる方法に関するこの回答は、インバウンドケースに動的に登録する方法を見つけました...

それで、これは行く方法ですか？ヘルプ/ガイダンスをいただければ幸いです。何日も DSL コードを調べ、ドキュメントを読んだ後、動的 ftp の例を実装し、それを FeedEntryMessageSource で動作するように適応させようと思います...その場合、私の質問は...その動的 ftp の例は機能しますXML構成を使用しますが、Java構成またはJava DSLのいずれかでそれを行うことは可能ですか?

アップデート

次のようにソリューションを実装しました。

DynamicFeedAdapter.java

そして、これはIFで機能し、 で定義された URL の1 つがある場合にのみ機能します。そうしないと、「プロパティ {feed.url} を解決できません」というメッセージが表示されて失敗します。そこで何が起こっているのかと思うのは、すべてで定義された sが熱心に初期化されるため、メイン メソッドの for ループで手動で作成された Bean (feed.url プロパティが注入されているため正常に動作する) を除いて、熱心に初期化された迷子のシングルトンであり、application.properties で定義された feed.url がない場合、プロパティを解決できず、すべてがうまくいきません。Springについて私が知っていることから、Beanを初期化できるはずだとわかっていますapplication.propertiesfeed.url=[insert url here]@BeanDynamicFeedAdapter.java@LazyDynamicFeedAdapter.javaそのため、この望ましくない漂遊シングルトンの問題児に巻き込まれることはありません。問題は今です...マークしただけでfeedAdapter() @Lazyは、Beanは初期化されません。自分で初期化するにはどうすればよいですか?

更新 - 問題は解決しました

テストしていないので、問題は、ブートがコンポーネント スキャン中に DynamicFeedAdapter を見つけていることだと思います。簡単な解決策は、兄弟パッケージに移動することです。MonsterFeedApplication が com.acme.foo にある場合は、アダプタ構成クラスを com.acme.bar に配置します。そうすれば、ブートはそれをアプリケーションの「一部」と見なしません

これは確かに問題でした。Gary の提案を実装した後、すべてが完璧に機能します。

WatchServiceDirectoryScanner/RecursiveLeafOnlyDirectoryScannerファイルシステムでファイルを処理するために使用しています。ファイル イベントが生成され、定義されたエンドポイントでメッセージが受信されますが、そのディレクトリ内のすべてのファイルを処理できない場合があります。

たとえば、15 個のファイルがある場合、10 個のファイルを処理する場合もあれば、5 個のファイルを処理する場合もあります。metastore には、15 個のファイルすべてに関する情報が含まれています"metadata-store.properties"。

spring-integration-configuration.xml:

BatchJobScheduler:

構成またはコードが不足しているようです。

提案に基づく:

完全なspring-integration.xmlログ ファイルを git リポジトリに配置しました - https://github.com/chandaku/spring-integration-issue。

ここでは 5 つのチャネルが作成され、同時に動作しています。一度に 1 つのチャネルのみを実行すると、ディレクトリ内のすべてのファイルが正常に処理されると思いますが、すべてのファイル チャネルを開くと問題が発生します。