Package nablarch.integration.messaging.wmq.provider

Class WmqMessagingProvider

java.lang.Object
nablarch.integration.messaging.wmq.provider.WmqMessagingProvider
All Implemented Interfaces:
Initializable, MessagingProvider
public class WmqMessagingProvider extends Object implements MessagingProvider, Initializable
WebSphere MQを使用したMessagingProviderの実装クラス。

  • Constructor Details

    • WmqMessagingProvider

      public WmqMessagingProvider()

  • Method Details

    • initialize

      public void initialize()
      WebSphere MQの初期化処理を行う。

      下記の処理を行う。

      • checkPoisonSetting()メソッドを呼び出し退避キューの設定不備がないことをチェックする。
      • 接続モード(CMQC.TRANSPORT_PROPERTY)をバインディングモードに設定する。
      • スレッド類縁性(CMQC.THREAD_AFFINITY_PROPERTY)にuseXaプロパティの値を設定する。
      • useProductSystemErrorOutputプロパティがfalseの場合はWebSphere MQによる標準エラー出力を無効化する。
      • 接続モード(CMQC.TRANSPORT_PROPERTY)をバインディングモードに設定する。
      • poisonQueueNamePatternが指定された場合はreceivedQueueNameを使用してフォーマットした退避キュー名を設定する。
      Specified by:
      initialize in interface Initializable

    • checkPoisonSetting

      protected void checkPoisonSetting()
      退避キューの設定不備がないことをチェックする。

      チェック内容は下記のとおり。

      • receivedQueueNameが指定されない場合は、 poisonQueueNamePatternとpoisonQueueNameも指定されないこと。
      • receivedQueueNameが指定された場合は、 poisonQueueNamePatternとpoisonQueueNameの両方が指定されないか、 いずれか一方のみ指定されること。
      設定不備が見つかった場合は実行例例外を送出する。

    • createContext

      public MessagingContext createContext()
      WmqMessagingContextを生成する。 
       指定された設定情報からMQQueueManagerを生成し、
 MQQueueマップ(送信先と受信先とポイズンメッセージ送信先)の初期化を行う。
 
 MQQueueManagerの生成はcreateMQQueueManager()メソッドに委譲する。
 
 送信先MQQueueマップの初期化はgetMQQueues(MQQueueManager, Collection, int)メソッド、
 受信先MQQueueとポイズンメッセージ送信先MQQueueの初期化はgetPoisonQueueOpenOptions()メソッドに委譲する。
 
 送信先MQQueueのオープンを制御するオプションはgetSendingQueueOpenOptions()メソッド、
 受信先MQQueueのオープンを制御するオプションはgetReceivedQueueOpenOptions()メソッド、
 ポイズンメッセージ送信先MQQueueのオープンを制御するオプションはgetPoisonQueueOpenOptions()メソッド、
 から取得する。
 
 MQExceptionを捕捉した場合は、
 MessagingExceptionFactory.createMessagingException(String, Throwable) メソッド
 に例外処理を委譲する。
      Specified by:
      createContext in interface MessagingProvider
      Returns:
      WmqMessagingContext

    • createMQQueueManager

      protected com.ibm.mq.MQQueueManager createMQQueueManager() throws com.ibm.mq.MQException
      キューマネージャ名称を指定してMQQueueManagerを生成する。
      Returns:
      MQQueueManager
      Throws:
      com.ibm.mq.MQException - MQQueueManagerの生成に失敗した場合

    • getSendingQueueOpenOptions

      protected int getSendingQueueOpenOptions()
      送信先MQQueueのオープンを制御するオプションを取得する。

      下記オプションを指定した値を返す。

      • CMQC.MQOO_OUTPUT
      • CMQC.MQOO_FAIL_IF_QUIESCING
      Returns:
      送信先MQQueueのオープンを制御するオプション

    • getReceivedQueueOpenOptions

      protected int getReceivedQueueOpenOptions()
      受信先MQQueueのオープンを制御するオプションを取得する。

      下記オプションを指定した値を返す。

      • CMQC.MQOO_INPUT_SHARED
      • CMQC.MQOO_FAIL_IF_QUIESCING
      Returns:
      受信先MQQueueのオープンを制御するオプション

    • getPoisonQueueOpenOptions

      protected int getPoisonQueueOpenOptions()
      ポイズンメッセージの退避先MQQueueのオープンを制御するオプションを取得する。

      getSendingQueueOpenOptions()メソッドに処理を委譲する。

      Returns:
      ポイズンメッセージの退避先MQQueueのオープンを制御するオプション

    • getMQQueues

      protected Map<String,com.ibm.mq.MQQueue> getMQQueues(com.ibm.mq.MQQueueManager mqQueueManager, Collection<String> queueNames, int openOptions) throws com.ibm.mq.MQException
      指定されたキュー名とオプションを使用して、正常にオープンされたMQQueueを取得する。
      Parameters:
      mqQueueManager - MQQueueManager
      queueNames - キュー名リスト
      openOptions - MQQueueのオープンを制御するオプション
      Returns:
      正常にオープンされたMQQueueマップ(キーはキュー名)
      Throws:
      com.ibm.mq.MQException - MQQueueのオープンに失敗した場合

    • getMQQueue

      protected com.ibm.mq.MQQueue getMQQueue(com.ibm.mq.MQQueueManager mqQueueManager, String queueName, int openOptions) throws com.ibm.mq.MQException
      指定されたキュー名とオプションを使用して、正常にオープンされたMQQueueを取得する。
      Parameters:
      mqQueueManager - MQQueueManager
      queueName - キュー名
      openOptions - MQQueueのオープンを制御するオプション
      Returns:
      正常にオープンされたMQQueue。キュー名の指定がない場合はnull
      Throws:
      com.ibm.mq.MQException - MQQueueのオープンに失敗した場合

    • putMessage

      protected String putMessage(com.ibm.mq.MQQueue mqQueue, SendingMessage sendingMessage)
      送信メッセージを指定されたMQQueueに書き込む。

      送信メッセージからMQMessageを作成しMQQueueに書き込む。
      メッセージ送信前のMQMDフィールドの設定は、 WmqMqmdFieldsOperator.setFieldsBeforeSend(SendingMessage, MQMessage, long)メソッドに委譲する。
      MQQueueへの書き込みを制御するオプションはgetGetMessageOptions(String, long)メソッドから取得する。
      メッセージ送信後のMQMDフィールドの取得は、 WmqMqmdFieldsOperator.getFieldsAfterSend(MQMessage, SendingMessage)メソッドに委譲する。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッドに例外処理を委譲する。
      Parameters:
      mqQueue - MQQueue
      sendingMessage - 送信メッセージ
      Returns:
      メッセージID

    • getPutMessageOptions

      protected com.ibm.mq.MQPutMessageOptions getPutMessageOptions()
      MQQueueへのメッセージ書き込みを制御するオプションを取得する。

      下記オプションを指定した値を返す。

      Returns:
      MQQueueへのメッセージ書き込みを制御するオプション

    • getPutPoisonMessageOptions

      protected com.ibm.mq.MQPutMessageOptions getPutPoisonMessageOptions()
      MQQueueへのポイズンメッセージ書き込みを制御するオプションを取得する。

      getPutMessageOptions()に処理を委譲する。

      Returns:
      MQQueueへのメッセージ書き込みを制御するオプション

    • getPutSyncpointOption

      protected int getPutSyncpointOption()
      メッセージ書き込み時に使用する同期点オプションを取得する。

      useXaプロパティがtrueの場合はCMQC.MQPMO_SYNCPOINT、 falseの場合はCMQC.MQPMO_NO_SYNCPOINTを返す。

      Returns:
      メッセージ書き込み時に使用する同期点オプション

    • getMessage

      protected ReceivedMessage getMessage(com.ibm.mq.MQQueue receivedMqQueue, String messageId, long timeout, com.ibm.mq.MQQueue poisonMqQueue)
      指定されたMQQueueから受信メッセージを読み込む。

      MQQueueから読み込んだMQMessageから受信メッセージを作成する。
      読み込んだMQMessageがポイズンメッセージの場合は processPoisonMessage(MQQueue, MQMessage)メソッドに処理を委譲する。 MQMessageがポイズンメッセージであるか否かの判定は isPoisonMessage(MQMessage)メソッドが行う。

      メッセージ受信前のMQMDフィールドの設定は WmqMqmdFieldsOperator.setFieldsBeforeReceive(String, MQMessage)メソッドに委譲する。
      MQQueueへの書き込みを制御するオプションはgetGetMessageOptions(String, long)メソッドから取得する。
      メッセージ受信後のMQMDフィールドの取得は WmqMqmdFieldsOperator.getFieldsAfterReceive(MQMessage, ReceivedMessage)メソッドに委譲する。

      MQExceptionが送出され、理由コードがCMQC.MQRC_NO_MSG_AVAILABLEの場合はnullを返す。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッド に例外処理を委譲する。

      Parameters:
      receivedMqQueue - MQQueue
      messageId - 相関IDに指定するメッセージID
      timeout - タイムアウト値(単位:msec)
      poisonMqQueue - ポイズンメッセージ送信先MQQueue。指定がない場合はnull
      Returns:
      受信メッセージ。受信できなかった場合はnull

    • isPoisonMessage

      protected boolean isPoisonMessage(com.ibm.mq.MQMessage mqMessage)
      指定されたMQMessageがポイズンメッセージであるか否かを判定する。

      MQMessageのバックアウト回数がbackoutLimitプロパティより大きい場合にtrueを返す。

      Parameters:
      mqMessage - MQMessage
      Returns:
      ポイズンメッセージである場合はtrue

    • processPoisonMessage

      protected void processPoisonMessage(com.ibm.mq.MQQueue poisonMqQueue, com.ibm.mq.MQMessage mqMessage) throws ProcessAbnormalEnd, MessagingException
      ポイズンメッセージの処理を行う。 
       ポイズンメッセージ送信先MQQueueの指定有無により処理が分かれる。
 
 ポイズンメッセージ送信先MQQueueの指定がない場合
 
   メッセージ受信処理が無限ループになるのを防止するために、
   ProcessAbnormalEndを送出しプロセスを異常終了させる。
   ProcessAbnormalEndの終了コードと障害コードには、
   backoutLimitExceededExitCodeプロパティとbackoutLimitExceededFailureCodeプロパティを指定する。
 
 ポイズンメッセージ送信先MQQueueが指定された場合
 
   FATALレベルで障害ログを出力し、指定されたMQQueueにポイズンメッセージを送信する。
   送信時は、メッセージ存続時間を無制限、元々設定されていたメッセージIDを相関IDに設定する。
   ポイズンメッセージの送信でMQExceptionが送出された場合は、
   送出されたMQExceptionがリトライ可能な例外である場合は再送出し、
   リトライ不可の場合は、メッセージ受信処理が無限ループになるのを防止するために、
   ProcessAbnormalEndを送出しプロセスを異常終了させる。
   ProcessAbnormalEndの終了コードと障害コードには、
   backoutLimitExceededExitCodeプロパティとbackoutLimitExceededFailureCodeプロパティを指定する。
      Parameters:
      poisonMqQueue - ポイズンメッセージ送信先MQQueue。指定がない場合はnull
      mqMessage - ポイズンメッセージ
      Throws:
      ProcessAbnormalEnd - ポイズンメッセージ送信先MQQueueの指定がない場合。 ポイズンメッセージの送信に失敗し、かつリトライ不可な場合
      MessagingException - ポイズンメッセージの送信に失敗し、かつリトライ可能な場合

    • getGetMessageOptions

      protected com.ibm.mq.MQGetMessageOptions getGetMessageOptions(String messageId, long timeout)
      MQQueueからのメッセージ読み取りを制御するオプションを取得する。

      下記オプションを指定した値を返す。

      • getGetSyncpointOption()メソッドが返す同期点オプション
      • CMQC.MQPMO_NEW_MSG_ID
      • CMQC.MQGMO_WAIT
      • 相関IDに指定するメッセージIDが指定された場合はCMQC.MQMO_MATCH_CORREL_ID、 指定がない場合はCMQC.MQMO_NONE
      • タイムアウト値。タイムアウト値が0以下の場合はdefaultResponseTimeoutプロパティの値を使用する。
      Parameters:
      messageId - 相関IDに指定するメッセージID
      timeout - タイムアウト値(単位:msec)
      Returns:
      MQQueueからのメッセージ読み取りを制御するオプション

    • getGetSyncpointOption

      protected int getGetSyncpointOption()
      メッセージ読み取り時に使用する同期点オプションを取得する。

      useXaプロパティがtrueの場合はCMQC.MQPMO_SYNCPOINT、 falseの場合はCMQC.MQPMO_NO_SYNCPOINTを返す。

      Returns:
      メッセージ読み取り時に使用する同期点オプション

    • disconnect

      protected void disconnect(com.ibm.mq.MQQueueManager mqQueueManager)
      MQQueueManagerの接続を切断する。

      MQExceptionが送出された場合はTRACEレベルのログを出力し、MQExceptionを送出しない。

      Parameters:
      mqQueueManager - MQQueueManager

    • close

      protected void close(Map<String,com.ibm.mq.MQQueue> mqQueues)
      指定された全てのMQQueueをクローズする。

      個々のMQQueueを指定してclose(MQQueue)を呼び出す。

      Parameters:
      mqQueues - MQQueueマップ

    • close

      protected void close(com.ibm.mq.MQQueue mqQueue)
      指定されたMQQueueをクローズする。

      MQExceptionが送出された場合はTRACEレベルのログを出力し、MQExceptionを送出しない。

      Parameters:
      mqQueue - MQQueue

    • getJdbcConnection

      protected Connection getJdbcConnection(com.ibm.mq.MQQueueManager mqQueueManager, XADataSource xaDataSource) throws SQLException, XAException
      MQQueueManagerからConnectionを取得する。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッド に例外処理を委譲する。

      Parameters:
      mqQueueManager - MQQueueManager
      xaDataSource - XAデータソース
      Returns:
      Connection
      Throws:
      SQLException - Connectionの取得に失敗した場合
      XAException - Connectionの取得に失敗した場合

    • checkXa

      protected void checkXa()
      分散トランザクションを使用するに設定されていることをチェックする。

      useXaプロパティがfalseの場合はIllegalStateExceptionを送出する。

    • begin

      protected void begin(com.ibm.mq.MQQueueManager mqQueueManager)
      分散トランザクションを開始する。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッド に例外処理を委譲する。

      Parameters:
      mqQueueManager - MQQueueManager

    • commit

      protected void commit(com.ibm.mq.MQQueueManager mqQueueManager)
      分散トランザクションをコミットする。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッド に例外処理を委譲する。

      Parameters:
      mqQueueManager - MQQueueManager

    • backout

      protected void backout(com.ibm.mq.MQQueueManager mqQueueManager)
      分散トランザクションをバックアウトする。

      MQExceptionを捕捉した場合は、 MessagingExceptionFactory.createMessagingException(String, Throwable)メソッド に例外処理を委譲する。

      Parameters:
      mqQueueManager - MQQueueManager

    • setQueueManagerName

      public WmqMessagingProvider setQueueManagerName(String queueManagerName)
      キューマネージャ名称を設定する。
      Parameters:
      queueManagerName - キューマネージャ名称
      Returns:
      このオブジェクト自体

    • setSendingQueueNames

      public WmqMessagingProvider setSendingQueueNames(List<String> sendingQueueNames)
      送信キュー名(論理名)リストを設定する。
      Parameters:
      sendingQueueNames - 送信キュー名(論理名)リスト
      Returns:
      このオブジェクト自体

    • setReceivedQueueName

      public WmqMessagingProvider setReceivedQueueName(String receivedQueueName)
      受信キュー名(論理名)を設定する。
      Parameters:
      receivedQueueName - 受信キュー名(論理名)
      Returns:
      このオブジェクト自体

    • setPoisonQueueNamePattern

      public void setPoisonQueueNamePattern(String poisonQueueNamePattern)
      退避キュー名パターン(論理名)を設定する。 
       受信先キュー名を使用した命名規約により退避キュー名が規定されている場合は、
 本プロパティを指定する。
 本プロパティの設定例を示す。
 
     設定値: "%s.POISON"
     受信先キュー名が"TEST"の場合は、"TEST.POISON"となる。
      Parameters:
      poisonQueueNamePattern - 退避キュー名パターン(論理名)

    • setPoisonQueueName

      public void setPoisonQueueName(String poisonQueueName)
      退避キュー名(論理名)を設定する。 
       退避キュー名が受信先キュー名と命名規約により規定されていない場合は、
 本プロパティを指定する。
      Parameters:
      poisonQueueName - 退避キュー名(論理名)

    • setBackoutLimit

      public WmqMessagingProvider setBackoutLimit(int backoutLimit)
      バックアウト回数の上限値を設定する。

      デフォルトは0。

      Parameters:
      backoutLimit - バックアウト回数の上限値
      Returns:
      このオブジェクト自体

    • setBackoutLimitExceededExitCode

      public WmqMessagingProvider setBackoutLimitExceededExitCode(int backoutLimitExceededExitCode)
      バックアウト上限を超えた場合に使用する終了コード(プロセスを終了(System.exit(int))する際に設定する値)を設定する。

      デフォルトは190。

      Parameters:
      backoutLimitExceededExitCode - バックアウト上限を超えた場合に使用する終了コード(プロセスを終了(System.exit(int))する際に設定する値)
      Returns:
      このオブジェクト自体

    • setBackoutLimitExceededFailureCode

      public WmqMessagingProvider setBackoutLimitExceededFailureCode(String backoutLimitExceededFailureCode)
      バックアウト上限を超えた場合に使用する障害コードを設定する。
      Parameters:
      backoutLimitExceededFailureCode - バックアウト上限を超えた場合に使用する障害コード
      Returns:
      このオブジェクト自体

    • setPutPoisonFailedExitCode

      public void setPutPoisonFailedExitCode(int putPoisonFailedExitCode)
      ポイズンメッセージの送信に失敗した場合に使用する終了コード(プロセスを終了(System.exit(int))する際に設定する値)を設定する。

      デフォルトは191。

      Parameters:
      putPoisonFailedExitCode - ポイズンメッセージの送信に失敗した場合に使用する終了コード(プロセスを終了(System.exit(int))する際に設定する値)

    • setPutPoisonFailedFailureCode

      public void setPutPoisonFailedFailureCode(String putPoisonFailedFailureCode)
      ポイズンメッセージの送信に失敗した場合に使用する障害コードを設定する。
      Parameters:
      putPoisonFailedFailureCode - ポイズンメッセージの送信に失敗した場合に使用する障害コード

    • setDefaultResponseTimeout

      public MessagingProvider setDefaultResponseTimeout(long defaultResponseTimeout)
      デフォルトタイムアウト値(単位:msec)を設定する。

      デフォルトは300秒。

      Specified by:
      setDefaultResponseTimeout in interface MessagingProvider
      Parameters:
      defaultResponseTimeout - デフォルトタイムアウト値(単位:msec)
      Returns:
      このオブジェクト自体

    • setDefaultTimeToLive

      public MessagingProvider setDefaultTimeToLive(long defaultTimeToLive)
      送信電文の有効期間(単位:msec)を設定する。

      デフォルトは60秒。

      Specified by:
      setDefaultTimeToLive in interface MessagingProvider
      Parameters:
      defaultTimeToLive - 送信電文の有効期間(単位:msec)
      Returns:
      このオブジェクト自体

    • setMqmdFieldsOperator

      public WmqMessagingProvider setMqmdFieldsOperator(WmqMqmdFieldsOperator mqmdFieldsOperator)
      WmqMqmdFieldsOperatorオブジェクトを設定する。

      デフォルトはBasicWmqMqmdFieldsOperator。

      Parameters:
      mqmdFieldsOperator - WmqMqmdFieldsOperatorオブジェクト
      Returns:
      このオブジェクト自体

    • setMessagingExceptionFactory

      public WmqMessagingProvider setMessagingExceptionFactory(MessagingExceptionFactory messagingExceptionFactory)
      MessagingExceptionファクトリオブジェクトを設定する。

      デフォルトはBasicWmqMessagingExceptionFactory。

      Specified by:
      setMessagingExceptionFactory in interface MessagingProvider
      Parameters:
      messagingExceptionFactory - MessagingExceptionファクトリオブジェクト
      Returns:
      このオブジェクト自体

    • setUseXa

      public void setUseXa(boolean useXa)
      分散トランザクションを使用するか否かを設定する。

      デフォルトはtrue。

      Parameters:
      useXa - 分散トランザクションを使用する場合はtrue

    • setUseProductSystemErrorOutput

      public void setUseProductSystemErrorOutput(boolean useProductSystemErrorOutput)
      WebSphere MQによるMQException発生時の標準エラー出力を使用するか否かを設定する。

      デフォルトはfalse。

      Parameters:
      useProductSystemErrorOutput - WebSphere MQによるMQException発生時の標準エラー出力を使用する場合はtrue