9.4. 例

Saga の完了時には、何らかの処理が必要となります。何らかの問題が発生して Saga がキャンセルされた場合には、補正エンドポイントが呼び出されます。Saga が正常に完了した場合は、完了エンドポイント を呼び出して追加の処理を行うことができます。たとえば、上記の注文サービスでは、実際に注文の準備を開始するために、注文がいつ完了したか (そしてクレジット (残高) が予約されたか) を知る必要があることがあります。支払いが完了していないのに、注文の準備を開始したくはありません (最新の CPU のように、読み込み権限を確認する前に予約メモリーへのアクセス権を与えてしまうのとは異なります) 。これは、direct:newOrder エンドポイントを変更すると簡単に実現できます。

from("direct:newOrder")
  .saga()
  .propagation(SagaPropagation.MANDATORY)
  .compensation("direct:cancelOrder")
  .completion("direct:completeOrder")
    .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
    .bean(orderManagerService, "newOrder")
    .log("Order ${body} created");

from("direct:completeOrder")
  .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
  .bean(orderManagerService, "findExternalId")
  .to("jms:prepareOrder")
  .log("Order ${body} sent for preparation");

Saga が完了すると、注文は準備のために JMS キューに送信されます。補正アクションと同様、完了アクションも Saga のコーディネーターによって複数回呼び出される可能性があります (特にネットワークエラーなどのエラーが発生した場合) 。この例では、prepareOrder JMS キューをリッスンするサービスが重複を受け取る可能性を備えています (重複をどのように扱うかの例は、Idempotent Consumer EIP を参照してください)。

Saga のいくつかのオプションを使用してカスタム識別子を登録することができます。たとえば、クレジット (残高) サービスは以下のようにリファクタリングされます。

from("direct:reserveCredit")
  .bean(idService, "generateCustomId")
  .to("direct:creditReservation")

from("direct:creditReservation")
  .saga()
  .propagation(SagaPropagation.SUPPORTS)
  .option("CreditId", body())
  .compensation("direct:creditRefund")
    .bean(creditService, "reserveCredit")
    .log("Credit ${header.amount} reserved. Custom Id used is ${body}");

from("direct:creditRefund")
  .transform(header("CreditId")) // retrieve the CreditId option from headers
  .bean(creditService, "refundCredit")
  .log("Credit for Custom Id ${body} refunded");

direct:creditReservation エンドポイントは、伝搬モードを SUPPORTS に設定することで Saga の外から呼び出すことができます。このようにして、複数のオプションを 1 つの Saga ルートの中で宣言できます。

Saga EIP でタイムアウトを設定することで、マシン障害の発生時に Saga が永久に停止したままにならないことが保証されます。Saga EIP の実装では、明示的にタイムアウトを指定していないすべての Saga EIP に対してデフォルトのタイムアウトが設定されます。タイムアウトの期限が切れると、Saga EIP はそれ以前に異なる決定がなされていない限り、Saga のキャンセル (およびすべての参加者への補正) を決定します。

タイムアウトは、Saga の参加者に対して以下のように設定できます。

from("direct:newOrder")
  .saga()
  .timeout(1, TimeUnit.MINUTES) // newOrder requires that the saga is completed within 1 minute
  .propagation(SagaPropagation.MANDATORY)
  .compensation("direct:cancelOrder")
  .completion("direct:completeOrder")
    // ...
    .log("Order ${body} created");

すべての参加者 (クレジット (残高) サービス、注文サービスなど) は、それぞれ独自のタイムアウトを設定できます。これらの参加者が 1 つに構成されたときに、それらのタイムアウトの中の最小値が Saga のタイムアウトとなります。タイムアウトは、以下のように Saga レベルで指定することもできます。

from("direct:buy")
  .saga()
  .timeout(5, TimeUnit.MINUTES) // timeout at saga level
    .to("direct:newOrder")
    .to("direct:reserveCredit");

上記の例では、MANDATORY および SUPPORTS の伝搬モードを使用していますが、他に何も指定されなかった場合に使用されるデフォルトの伝播モードである REQUIRED もあります。これらの伝搬モードは、トランザクションの文脈で使用される同等のモードと 1 対 1 に対応します。

Saga をすべて同期的に実行できず、たとえば非同期通信チャネルを使用した外部サービスとの通信などが必要となる場合には、完了モードを AUTO (デフォルト) には設定できません。これは、Saga を作成したエクスチェンジが完了した時点ではその Saga は完了していないためです。実行期間が長い (数時間、数日) Saga EIP では、このようなことがよくあります。このような場合には、MANUAL 完了モードを使用する必要があります。

from("direct:mysaga")
  .saga()
  .completionMode(SagaCompletionMode.MANUAL)
  .completion("direct:finalize")
  .timeout(2, TimeUnit.HOURS)
    .to("seda:newOrder")
    .to("seda:reserveCredit");

seda:newOrder および seda:reserveCredit に非同期の処理を追加します。これらは seda:operationCompleted に非同期のコールバックを送信します。

from("seda:operationCompleted") // an asynchronous callback
  .saga()
  .propagation(SagaPropagation.MANDATORY)
    .bean(controlService, "actionExecuted")
    .choice()
      .when(body().isEqualTo("ok"))
        .to("saga:complete") // complete the current saga manually (saga component)
    .end()

direct:finalize エンドポイントを追加することで、最終のアクションを実行できます。

完了モードを MANUAL に設定すると、ルート direct:mysaga でエクスチェンジが処理されても Saga が完了せずに持続します (最大持続時間は 2 時間に設定されます)。非同期アクションが両方とも完了することで、Saga が完了します。完了の呼び出しは、Camel Saga コンポーネントの saga :complete エンドポイントを使用して行われます。手動で Saga を補正するための同様のエンドポイントがあります (saga:compensate) 。