プラットフォームイベント(Platform Events)を試す

Summer '17で正式リリースされたプラットフォームイベントを試してみました.

プラットフォームイベントとは
  • Pub/Subのメッセージングシステム (e.g., Apache Kafka, Cloud Pub/Sub)
  • メッセージの配信者(Publisher)は,購読者(Subscriber)を意識せずメッセージを送信できる.そのため,コンポーネント同士が疎結合になる点がメリット.
  • プラットフォームイベントはImmutableであり,クエリ不可.
  • プラットフォームイベントは配信してから24時間レコードが保持される.
  • ちなみに,標準オブジェクトのEvent(行動)とは無関係.
プラットフォームイベントの登録
  • [設定] - [プラットフォームイベント]
  • カスタムオブジェクトのように,プラットフォームイベントはAPI参照名の末尾に__eが付与される.
  • 受信者は,replayID(再実行ID)を利用して過去のイベントを取得可.

f:id:shunkosa:20170601173052p:plain

プラットフォームイベントの配信(Publisher: Salesforce Internal)
  • Apexでは,EventBus.publish()メソッドでイベントを配信できる.
  • EventBus.publish()メソッドの呼び出しは,レコードのInsertと同等のため,通常のDMLと同じガバナ制約が適用される.また,このメソッドはList<Database.SaveResult>を返す.
//ケースがエスカレーションされていたらイベントを配信するトリガ
trigger CaseTrigger on Case (after update) {
    List<Case_Event__e> CaseEvents = new List<Case_Event__e>();
    for(Case c : Trigger.new)
        if(c.isEscalated){
            CaseEvents.add(
            	new Case_Event__e(Type__c = c.Type, 
                                  CaseNumber__c = c.CaseNumber, 
                                  Description__c = c.Description
                                  )
            );
        }
    }
    EventBus.publish(CaseEvents);
}
プラットフォームイベントの購読(Subscriber: External)

Force.comのGithubCometDプロトコルを実装したJavaのクライアントが公開されています.まずは,これを用いてイベントを購読してみます.

  • exampleパッケージ内のLoginExampleを実行.コマンドライン引数に ログインID パスワード トピック を順に指定
  • イベントを購読する際のトピックのフォーマットは /event/(イベントのAPI参照名)

下記は,先のトリガを有効化し,ケースをエスカレーションした際のJava側の標準出力例.うまく購読できています.

{schema=X1gI8VTM2N-J2F6sVaOWVA, 
 payload={
	CaseNumber__c=00001026, 
	CreatedById=005B0000000P1Vz, 
	Description__c=Test Description, 
	Type__c=Electrical, 
	CreatedDate=2017-06-01T07:39:02Z},
 event={replayId=1}}
プラットフォームイベントの配信(Publisher: External)
  • 通常のsObjectのレコード挿入と同様にAPI経由でイベントを配信(=イベントレコードを挿入).

先の例だと, /services/data/v40.0/sobjects/Case_Event__e/

プラットフォームイベントの購読(Subscriber: Internal Salesforce)
  • イベントのafter insertトリガを用いて,イベントを購読可能.
  • イベントレコードはImmutableであるため,beforeトリガやafter updateトリガには非対応.
trigger CaseEventTrigger on Case_Event__e (after insert) {
    for(Case_Event__e cEvent : Trigger.new){
    //例えば,ケース番号で元のケースをクエリして,元のケースを更新する等
    }
}
主な制約事項
  • 1時間あたりにpublishできるイベントレコード数の上限:100000 (DEは1000)
ポイント
  • イベントのPublish = イベントレコードのInsert