SalesforceのプラットフォームイベントとMuleSoftでSalesforceのデータをOracle DBに登録する

はじめに

近年、クラウドサービスや基幹システムなど、さまざまなシステムとデータを連携する機会が増加しているといわれています。
今回は、SalesforceのプラットフォームイベントとMuleSoftを使用したデータ連携の方法についてご紹介します。

プラットフォームイベントの作成

設定 > プラットフォームイベントからプラットフォームイベントを作成します。

Apexトリガーの作成

作成したプラットフォームイベントを発行するApexトリガーを作成します。
ステージが特定のものに変更された時のみプラットフォームイベントを発行します。

trigger OppTrigger on Opportunity (after update) {
    
    if(Trigger.isAfter && Trigger.isUpdate){
        OpportunityTriggerHandler.onAfterUpdate(Trigger.new, Trigger.oldMap);
    }
}

.

public without sharing class OpportunityTriggerHandler {
    
    private static final String STATUS_CLOSE = 'Closed Won';
    
    public static void onAfterUpdate(List<Opportunity> newRecords, Map<Id,Opportunity> oldRecordMap){
        List<ClosedOpportunity__e> targetList = new List<ClosedOpportunity__e>();
        
        for(Opportunity newRecord : newRecords){
            Opportunity oldRecord =  oldRecordMap.get(newRecord.Id);
            
            if(oldRecord.StageName != STATUS_CLOSE && newRecord.StageName == STATUS_CLOSE){
                ClosedOpportunity__e closedOppEvent = new ClosedOpportunity__e(
                    RecordId__c = newRecord.Id
                );
                targetList.add(closedOppEvent);
            }
        }
        
        if(targetList != null && targetList.size() > 0){
            List<Database.SaveResult> res = EventBus.publish(targetList);
        }
    }
}

.

Oracle DBにテーブルの作成

データを登録するテーブルを作成します。
※Oracle DBのインストール方法については割愛します。

CREATE TABLE OPPORTUNITY(
  ID NUMBER GENERATED ALWAYS AS IDENTITY,
  OPPORTUNITY_NAME VARCHAR2(120),
  ACCOUNT_NAME VARCHAR2(80),
  AMOUNT NUMBER(18,0),
  CREATED_DATE TIMESTAMP,
  UPDATE_DATE TIMESTAMP
)

.

Muleフローの作成

Salesforceで発行したプラットフォームイベントをMuleSoftでサブスクライブし、Oracle DBに登録するフローを作成します。

プラットフォームイベントのサブスクライブ

Salesforceで発行したプラットフォームイベントをサブスクライブするためには、「Subscribe channel listener」コネクタを使用します。
「Connector configuration」に接続するSalesforceアカウントの認証情報を設定します。
設定を行うと「Streaming channel」の欄にその組織のプラットフォームイベントが表示されるようになります。
今回作成した「クローズ商談連携」を指定します。

Salesforceからデータの取得

「Query xml stream」コネクタを使用してSalesforceからデータを取得します。
プラットフォームイベントのデータ構造は下記のようになっています。
プラットフォームイベントにカスタム項目として作成した「RecordId__c」をもとにSalesforceからデータを取得します。

{
  "data": {
    "schema": "6t9Yj5himrLmp9KJXDB4YA",
    "payload": {
      "CreatedById": "0057F000002WkX5QAK",
      "RecordId__c": "0067F000017oDxNQAU",
      "CreatedDate": "2024-10-28T05:02:42.090Z"
    },
    "event": {
      "EventUuid": "ca8630e5-f948-44ae-a1c9-03b3aa8859ac",
      "replayId": 34378482,
      "EventApiName": "ClosedOpportunity__e"
    }
  },
  "channel": "/event/ClosedOpportunity__e"
}

.

「Parameters」を利用することでデータを動的に取得ができるようになります。
OpportunityIdというパラメータにプラットフォームイベントから取得した「RecordId__c」を設定しています。

データの変換

「Transform Message」コネクタを使用してSalesforceから取得したデータをJSON形式に変換します。
変換はMuleSoftのDataWeaveという言語を使用して行います。

Oracle DBへデータの登録

「Bulk Insert」コネクタを使用して、変換したJSONをもとにOracle DBにデータを登録します。
「Input Parameters」には変換したJSONを指定して、SQL文にはJSONのキーをバインドすることで値が設定されます。
「Connector configuration」に接続するDBの認証情報を設定します。