Mulesoft Anypoint コネクターでインテグレーション開発

by Cho 2015.06.15

こんにちは、Professional Serviceの Choです。
今回は、Mulesoft Anypointコネクターを用いたデータ統合についてご紹介します。

Anypointコネクター


MulesoftのAnypointコネクターは、統合アプリケーションを容易にサードパーティAPIとスタンダードインテグレーションプロトコル(HTTP, SMTP, FTP)を連携するため用意されているコネクターセットです。統合アプリケーションの中でコネクターを利用し、プロトコルとAPIを使ってデータの送受信を行います。
Anypoint Studioにはあらかじめ30種類のコネクターが含まれていますが、Anypoint エクスチェンジで120以上のコネクターライブラリーから追加ダウンロードすることも可能です。開発者はRESTとSOAP APIのコネクターを作成できるだけでなく、AnypointコネクターDevKitを使って新しいコネクターを開発することもできます。

コネクターのタイプ

すべてのコネクターはエンドポイントとしてメッセージをアプリケーション内部通信もしくは外部データソースと送受信を行います。
エンドポイントタイプ:インバウンドもしくはアウトバウンドエンドポイントだけとして使えます。例:FTPコネクター。
オペレーションタイプ:API、ウェブサービスやデータベースに特定なアクションを起動します。例:SFDCコネクター。

HTTP コネクター
HTTPコネクターは指定されたホスト、ポート及びアドレスにHTTPとHTTPSリクエストの送受信を行います。受信タイプはHTTPリクエストを待ち受けてHTTPレスポンスを返すことができます。送信タイプはHTTPリクエストを送信してレスポンスを受けることができます。このコネクターはRAML仕様と結合できまるだけでなく、OAuth認証機能も有しています。

データベース コネクター
データベースコネクターは一つのインターフェースですべてのJDBCドライバを利用できます。このコネクターを使って多数のSQLオペレーションが行えます、基本のCRUD操作はもちろん、ストアドプロシージャもできます。クエリーの構成はパラメータを使って動的に生成できますし、一つのリクエストにより複数のSQLリクエストが実行できます、さらにDDLもサポートされます。

Salesforce コネクター
Mule ESBでSalesforceのAPIを呼び出すにはSalesforceコネクターを使います。オペレーションタイプにより、SOAP API、REST API、Bulk API、Streaming APIがサポートされます。コネクターからのコールはHTTPS接続を利用してXML式のリクエスト/レスポンスを行います。リクエストに必要になるヘッダの設定やエラーハンドリング、認証などは、全てSalesforceコネクタによって処理されるので、データの統合に注力することができます。

Microsoft サービスバスコネクター
Microsoft サービスバスコネクターはオンプレミスのWindowsサービスバスやクラウドのAzureサービスバスと連携するコネクターです。このコネクターはAMQP 1.0プロトコルを利用してキュー、トピック、イベントハブとのコミュニケーションをサポートできます。加えて、組込みマネージメントAPIを利用すると、サービスバスオブジェクトを判断して動的に提供することも可能です。

コネクターの使用例
メッセージキューは古いpoint-to-point接続技術ですが、アプリケーション数が膨大なクラウド時代においては、さまざまなインタフェースをサポートするコストと複雑性が大幅に増加します。こうした場合にMuleESBと連携してメッセージキューにも一対一の統合から一対多に拡張することが可能です。

今回は、SalesforceとAzure Service Busの連携シナリオを想定して、Microsoft サービスバスコネクターを使ってキューの作成、監視、送信機能及びSalesforce コネクターを使ってChatter投稿機能を利用してみます。

流れとしては、先ずキューを作成します、そしてMuleアプリケーションがAzureキューを監視します、最後にキューにメッセージを投げで、このメッセージをSalesforce Chatterに投稿します。

1、Salesforce開発者環境をサインアップします。

2、Microsoft Azureをサインアップします。
Azureメニューからサービスバス機能を有効化します。

サービスバスの”接続情報”ボタンをクリックして、アクセス情報をコピーしておきます。

3、Mulesoft Anypointスタジオをインストールします。
インストールが完了すると、このように表示されます。

4、コネクターを追加
スタジオでツールバーの二番目”∞”ボタンをクリックします。Exchangeライブラリーに様々なコネクターやテンプレートが表示されますが、今回は”Microsoft Azure Service Bus Connector”を検索してインストールします。(Webからもインストールできます)

新しいMuleプロジェクトを作成した後、右側のコネクター欄に”Microsoft Service Bus”、”Salesforce”コネクターが見えます。

5、実装
ここからは実装です、”testqueue”という名前のキューに対して三つのフローを作成します。
• キュー監視とChatter投稿フロー:キューを監視して、キューに入ったメッセージを使ってChatterに投稿します。
• キュー作成フロー:キュー作成オペレーションを行います。
• キュー送信フロー:キュー送信オペレーションを行います。

キュー監視とChatter投稿フロー

目的:このフローは”testqueue”キューを監視して、メッセージが来ましたら内容を取ってSObjectに変換してChatterに投稿します。
実装:
サービスバスの設定を作ってnamespace、userName、passwordを先ほどステップ2で保存された接続情報を入れます。Operationに”Queue send”を設定し、Source Queueに”testqueue”を設定します。
一個Loggerを作ってコンソールにもメッセージを出力します、Messageに”#[payload.body]”を入れます。
そしてDataMapperを作って受け取った値をChatterのFeedItemオブジェクトにマッピングします。
body、type、parentId三つのInput argumentsを作って、FeedItemに同じ名前の項目にマッピングします、body argumentの式に”#[payload.body]”を入れます。

最後にSalesforceコネクターを追加します。Salesforce: Basic authenticationという設定を追加してログイン情報を入れます。
コネクターのOperationに”Create single”を選んでsObject Typeを”Feed Item”を選びます。
sObject Field Mappingのところ”sObject: From Message”を選んで”#[payload]”を入れます。

キュー送信フロー

目的:今回のデモをするため、キューにメッセージを送信するのはHTTPリクエストで行います。このフローはHTTPリクエストの”message”パラメータからメッセージBodyを読み取って、指定されたAzureサービスバスキューにメッセージを送信します。

実装:
HTTP設定を作ってPortに”8083″を設定し、Base Pathに”/NewQMessage”を設定します。
次にPayloadをByte ArrayからStringに変換します。HTTPコネクターがデフォルトByte Array式でpayloadを渡していますが、ここでStringとして利用するため、ここで変換ステップを加えます。
サービスバスコネクターはの監視コネクターの設定と同じ設定を使います。コネクターのOperationに”Queue send”を設定し、Destination Queueに”testqueue”を設定します、そしてbody-Referenceに”#[message.inboundProperties.’http.query.params’.message]”をいれます。

キュー作成フロー

目的:HTTP Getリクエストをコマンドにして”testqueue”キューを作成します、クエリーパラメーター”queuePath”をキューの名前になります。
実装:
新たにHTTPコネクター設定を作って、Pathに”/createqueue”を設定し、Portに”8081″を設定します。
新たにサービスバスコネクターの設定を作ってnamespace、userName、passwordを先ほどステップ2で保存された接続情報を入れます。
サービスバスのOperationに”Queue create”を設定し、Queue Pathに”#[message.inboundProperties[‘http.query.params’][‘queuePath’]]”をいれます。
HTTPリクエストのqueuePathパラメータから読み取って、キューの名前としてAzureサービスバスに新規キューのメッセージを送信します。
サービスバス仕様により返したメッセージはJSON式です、サービスバスコネクターがデフォルトObject式でpayloadを渡していますが、payloadをJSONに認識させるよ、最後にobject-to-json-transformerを追加します。

自動作成したXMLコンフィギュレーションは以下となります。

    <http:listener-config name="HTTP_Listener_Configuration" host="0.0.0.0" port="8081" doc:name="HTTP Listener Configuration"/>
    <http:listener-config name="HTTP_Listener_Configuration2" host="0.0.0.0" port="8083" basePath="/NewQMessage" doc:name="HTTP Listener Configuration"/>
    <microsoftservicebus:azureConfig name="Microsoft_Service_Bus__Azure_Service_Bus" namespace="${azure.namespace}" userName="${azure.key.name}" password="${azure.key}" doc:name="Microsoft Service Bus: Azure Service Bus"/>
    <microsoftservicebus:azureConfig name="Microsoft_Service_Bus__Azure_Service_Bus1" namespace="${azure.namespace}" userName="${azure.key.name}" password="${azure.key}" doc:name="Microsoft Service Bus: Azure Service Bus"/>
    <sfdc:config name="Salesforce__Basic_authentication" username="${sfdc.username}" password="${sfdc.password}" doc:name="Salesforce: Basic authentication"/>
    <data-mapper:config name="String_To_FeedItem" transformationGraphPath="string_to_feeditem.grf" doc:name="String_To_FeedItem"/>
    <flow name="Createqueue">
        <http:listener config-ref="HTTP_Listener_Configuration" path="/queue" doc:name="HTTP"/>
        <microsoftservicebus:queue-create config-ref="Microsoft_Service_Bus__Azure_Service_Bus" queuePath="#[message.inboundProperties['http.query.params']['queuePath']]" doc:name="Microsoft Service Bus">
        	<microsoftservicebus:queue-description enableBatchedOperations="false" maxSizeInMegabytes="5120"/>
        </microsoftservicebus:queue-create>
        <json:object-to-json-transformer doc:name="Object to JSON"/>
    </flow>
    <flow name="getQueue">
        <http:listener config-ref="HTTP_Listener_Configuration" path="/getq" doc:name="HTTP"/>
        <microsoftservicebus:queues-list config-ref="Microsoft_Service_Bus__Azure_Service_Bus" doc:name="Microsoft Service Bus"/>
        <json:object-to-json-transformer doc:name="Object to JSON"/>
    </flow>
    <flow name="sendQueue">
        <http:listener config-ref="HTTP_Listener_Configuration2" path="/" doc:name="HTTP"/>
        <byte-array-to-string-transformer doc:name="Byte Array to String"/>
        <microsoftservicebus:queue-send config-ref="Microsoft_Service_Bus__Azure_Service_Bus1" destinationQueue="testqueue" body-ref="#[message.inboundProperties.'http.query.params'.message]" doc:name="Microsoft Service Bus">
            <microsoftservicebus:header durable="true"/>
        </microsoftservicebus:queue-send>
    </flow>
    <flow name="receivequeue">
        <microsoftservicebus:queue-receive config-ref="Microsoft_Service_Bus__Azure_Service_Bus1" sourceQueue="testqueue" doc:name="Microsoft Service Bus (Streaming)"/>
        <logger message="#[payload.body]" level="INFO" doc:name="Logger"/>
        <data-mapper:transform config-ref="String_To_FeedItem" input-ref="#[message.outboundProperties["MULE_ENCODING"]]" doc:name="String To FeedItem">
            <data-mapper:input-arguments>
                <data-mapper:input-argument key="type">TextPost</data-mapper:input-argument>
                <data-mapper:input-argument key="body">#[payload.body]</data-mapper:input-argument>
                <data-mapper:input-argument key="parentId">00500;/data-mapper:input-argument>
            </data-mapper:input-arguments>
        </data-mapper:transform>
        <sfdc:create-single config-ref="Salesforce__Basic_authentication" type="FeedItem" doc:name="Salesforce">
            <sfdc:object ref="#[payload]"/>
        </sfdc:create-single>
    </flow>

6、検証
アプリケーションを実行は、Run As > Mule Applicationから開始します。
まずブラウザでキュー作成リクエストを投げます: http://localhost:8081/createqueue?queuePath=testqueue

Azureのコンソールから作成されたキュー情報は見えます。

このアプリケーションはキューの監視していますので、起動後のコンソールで受信待ちの状態が見えます。

キューができましたら実際のメッセージを送ります。
ブラウザでリクエストします。

“HelloMule~~~”はメッセージbodyの内容です。

メッセージが受信されました、コンソールで受信した内容を出力して、同じメッセージが見えます:

そしてSalesforceのChatterページから見るとこのメッセージが正しく投稿しました。

従来のインテグレーション開発では、先ず開発者がAzureのドキュメントを勉強して、Salesforceのドキュメントを勉強して、そしてサンプルコードを熟読してから実装し始めます、途中バグか発生するかもしれません。開発者スキルの要求が高いし、時間もコストもかなりかかります。外部アプリケーションがバージョンアップしたら、まだ繰り返し作業を始めます。

Mule Anypointコネクターを使うと、このようなヘビーな作業をMuleに任せて、開発者はコンフィギュレーションレベルの開発でポイント・アンド・クリックだけで短期間にデータ統合アプリケーションを作ることができます。

まとめ

AnypointコネクターはMuleと外部アプリケーション連携用のソフトウェアです、外部リソースはデータベース、プロトコル、APIを含めて任意のコンテンツです。Mulesoftは120以上のコネクターが用意されて、さらに要望に応じてコネクターDevKitを使って開発者がカスタマイズコネクターも作れます。Anypointコネクターの利用により、APIの複雑さに関わることなく、開発時間も月単位から週単位へと大幅に削減できます。

※パソナテキーラのMulesoft詳細ページはこちら

お問い合わせ

掲載記事に関するあらゆること、お気軽にお問い合わせください。ご相談は無料です。