PlatformEventに触れてみよう（Javaアプリでイベントを購読してみる）

trigger DemoOrderTrigger on DemoOrder__c (after update) {
    List<DemoOrderEvent__e> events = new List<DemoOrderEvent__e>();
    for(DemoOrder__c od : Trigger.new){
        if(!String.isEmpty(od.Status__c) && od.Status__c == '出荷依頼'){
            DemoOrderEvent__e ev = new DemoOrderEvent__e();
            ev.OrderNumber__c = od.Name;
            ev.Status__c = od.Status__c;
            events.add(ev);
        }
    }
    if(!events.isEmpty()){
        EventBus.publish(events);
    }
}

#!/bin/sh
keytool -genkeypair -alias keystore -keyalg RSA -keysize 2048 -keystore keystore.jks -storetype JKS -validity 90 -keypass keypassword -storepass storepassword

#!/bin/sh
keytool -v -export -file mycert.cer -keystore keystore.jks -storepass storepassword -alias keystore

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.gmail.jimaoka.platformevent</groupId>
  <artifactId>consumerexample</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>consumerexample</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
        <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.cometd.java/cometd-java-client -->
        <dependency>
            <groupId>org.cometd.java</groupId>
            <artifactId>cometd-java-client</artifactId>
            <version>3.1.2</version>
        </dependency>
  </dependencies>
</project>

package com.gmail.jimaoka.platformevent.consumerexample;

import com.gmail.jimaoka.platformevent.consumerexample.client.PlatformEventService;
import com.gmail.jimaoka.platformevent.consumerexample.oauth.OAuthResponse;
import com.gmail.jimaoka.platformevent.consumerexample.oauth.OAuthService;

public class App {
  public static void main( String[] args ){
    OAuthService oauthService = new OAuthService();
    try {
      String jwtBeareToken = oauthService.getJWTBeareToken();
      System.out.println("jwtBeareToken:" + jwtBeareToken);
      OAuthResponse oauthResponse = oauthService.getOAuthResponse(jwtBeareToken);
      System.out.println("access_token:" + oauthResponse.getAccess_token());
      PlatformEventService pevService = 
          new PlatformEventService(oauthResponse.getInstance_url(), oauthResponse.getAccess_token());
      pevService.start();
    } catch (Exception e) {
      System.out.println("exception");
      e.printStackTrace();
    }
  }
}

package com.gmail.jimaoka.platformevent.consumerexample.oauth;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.PrivateKey;
import java.security.Signature;
import java.security.SignatureException;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.text.MessageFormat;
import java.util.Base64;

import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.util.ssl.SslContextFactory;

import com.fasterxml.jackson.databind.ObjectMapper;

public class OAuthService {
  private static final String END_POINT = "https://login.salesforce.com/services/oauth2/token";
  private static final String AUDIENCE = "https://login.salesforce.com";
  // 接続アプリケーションのコンシュマー鍵の値を設定 
  private static final String CLIENT_ID = "3MVG9yZ.WNe6byQCanC7xsqg81Nfl068RgH1FBkktUix_dgDHv2wRVxxxxxxxxxxxxx";
  private static final String GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer";
  
  public String getJWTBeareToken() throws KeyStoreException, NoSuchAlgorithmException, CertificateException, FileNotFoundException, IOException, UnrecoverableKeyException, InvalidKeyException, SignatureException{
    String header = "{\"alg\":\"RS256\"}";
    String claimTemplate = "'{'\"iss\": \"{0}\", \"sub\": \"{1}\", \"aud\": \"{2}\", \"exp\": \"{3}\"'}'";
      
    StringBuffer token = new StringBuffer();
    token.append(Base64.getUrlEncoder().encodeToString(header.getBytes("UTF-8")));
    token.append(".");
      
    String[] claimArray = new String[4];
    claimArray[0] = CLIENT_ID;
    // 認証を行うSalesforceユーザID
    claimArray[1] = "hogehoge@gmail.com";
    claimArray[2] = AUDIENCE;
    claimArray[3] = Long.toString((System.currentTimeMillis() / 1000) + 300);
      
    MessageFormat claims = new MessageFormat(claimTemplate);
    String payload = claims.format(claimArray);
    token.append(Base64.getUrlEncoder().encodeToString(payload.getBytes("UTF-8")));

    KeyStore keystore = KeyStore.getInstance("JKS");
    // 鍵のパス
    keystore.load(new FileInputStream("./keystore/keystore.jks"), "storepassword".toCharArray());
    PrivateKey privateKey = (PrivateKey)keystore.getKey("keystore", "keypassword".toCharArray());
      
    Signature signature = Signature.getInstance("SHA256withRSA");
    signature.initSign(privateKey);
    signature.update(token.toString().getBytes("UTF-8"));
    String signedPayload = Base64.getUrlEncoder().encodeToString(signature.sign());
      
    token.append(".");
    token.append(signedPayload);
    return token.toString();
  }
  
  public OAuthResponse getOAuthResponse(String jwtBeareToken) throws Exception {
    SslContextFactory sslContextFactory = new SslContextFactory();
    HttpClient httpClient = new HttpClient(sslContextFactory);
    httpClient.start();
    ContentResponse response = httpClient.POST(END_POINT)
        .param("grant_type", GRANT_TYPE)
        .param("assertion", jwtBeareToken)
        .send();
    httpClient.stop();
    
    ObjectMapper objectMapper = new ObjectMapper();
    OAuthResponse oauthResponse = 
        objectMapper.readValue(response.getContentAsString(), OAuthResponse.class);
    return oauthResponse;
  }

  public OAuthResponse getOAuthResponse() throws  Exception {
    return getOAuthResponse(getJWTBeareToken());
  }
}

package com.gmail.jimaoka.platformevent.consumerexample.oauth;

public class OAuthResponse {
  private String access_token;
  private String scope;
  private String instance_url;
  private String token_type;
  private String id;
  
  public String getAccess_token() {
    return access_token;
  }
  public void setAccess_token(String access_token) {
    this.access_token = access_token;
  }
  public String getScope() {
    return scope;
  }
  public void setScope(String scope) {
    this.scope = scope;
  }
  public String getInstance_url() {
    return instance_url;
  }
  public void setInstance_url(String instance_url) {
    this.instance_url = instance_url;
  }
  public String getToken_type() {
    return token_type;
  }
  public void setToken_type(String token_type) {
    this.token_type = token_type;
  }
  public String getId() {
    return id;
  }
  public void setId(String id) {
    this.id = id;
  }
}

package com.gmail.jimaoka.platformevent.consumerexample.client;

import org.cometd.client.BayeuxClient;
import org.cometd.client.transport.LongPollingTransport;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.util.ssl.SslContextFactory;

public class PlatformEventService {
  private static final String COMETD_ENDPOINT = "/cometd/41.0";
  private static final String CHANNEL = "/event/DemoOrderEvent__e";

  private String instanceUrl;
  private String accessToken;
  private HttpClient httpClient;
  private BayeuxClient client;
  
  public PlatformEventService(String instanceUrl, String accessToken){
    this.instanceUrl = instanceUrl;
    this.accessToken = accessToken;
  }
  
  public void start() throws Exception{
    SslContextFactory sslContextFactory = new SslContextFactory();
    httpClient = new HttpClient(sslContextFactory);
    httpClient.start();
    
    LongPollingTransport httpTransport = new LongPollingTransport(null, httpClient){
      @Override
      protected void customize(Request request) {
        request.header("Authorization", "Beare " + accessToken);
      }
    };
    
    client = new BayeuxClient(instanceUrl + COMETD_ENDPOINT, httpTransport);
    client.handshake((channel, message) -> {
      if(message.isSuccessful()){
        System.out.println(message.toString());
      }
    });

    client.getChannel(CHANNEL).subscribe((channel, message) -> {
      System.out.println(message.toString());
    });
  }
  
  public void stop() throws Exception{
    if(client != null){
      client.disconnect();
    }
    if(httpClient != null){
      httpClient.stop();
    }
  }
}