SAP Event Meshを試してみる

お仕事でEvent Meshを触る機会があったので自分でも触ってみようと思いました

わかりやすい説明とかは書けないのでこちらの記事を読んで頂くとだいたいのことはわかると思います（感謝）

https://qiita.com/tami/items/727fe127158eea39fd3e

また実務での利用イメージはこちらを見ていただくと良いかと思います。

https://discovery-center.cloud.sap/protected/index.html#/missiondetail/3242/3268/?tab=overview

S/4のBusinessPartnerが更新されるとEventMeshにMQTTプロトコルで連携がされてODataで見に行くようになっています。

# 今回の流れ

https://github.com/SAP-samples/event-mesh-client-nodejs-samples

Node.jsのメッセージクライアントリポジトリがあるのでこちら使ってみます。
基本的にこちらの解説に従って進めていきます。
大きな流れとしては
１．準備
２．Queue Subscriptionsの確認
３．Webhook  Subscriptionsの確認

## 準備

### EventMeshのinstance作ったりServiceKeyの発行
上記のURLにも記載はありますがチュートリアルもあるのでこちらも参考に

https://developers.sap.com/tutorials/cp-enterprisemessaging-learn-messaging-concepts.html

### コード準備
```cmd
git clone https://github.com/SAP-samples/event-mesh-client-nodejs-samples.git
cd xb-msg-amqp-v100-samples
npm install
```

[cf-sample-config.js](https://github.com/SAP-samples/event-mesh-client-nodejs-samples/blob/main/xb-msg-amqp-v100-samples/config/cf-sample-config.js)をServiceKeyに合わせて修正

```ServiceKey.json
{
  "uaa": {
    "credential-type": "binding-secret",
    "clientid": "sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
    "clientsecret": "0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw=",
    "xsappname": "clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
    "url": "https://bb8713batrial.authentication.us10.hana.ondemand.com"
  },
  "xsappname": "clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
  "management": [
    {
      "oa2": {
        "clientid": "sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
        "clientsecret": "0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw=",
        "tokenendpoint": "https://bb8713batrial.authentication.us10.hana.ondemand.com/oauth/token",
        "granttype": "client_credentials"
      },
      "uri": "https://enterprise-messaging-hub-backend.cfapps.us10.hana.ondemand.com"
    }
  ],
  "serviceinstanceid": "63fe8d60-c2a5-4ecf-99b2-69e00ccbea62",
  "messaging": [
    {
      "oa2": {
        "clientid": "sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
        "clientsecret": "0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw=",
        "tokenendpoint": "https://bb8713batrial.authentication.us10.hana.ondemand.com/oauth/token",
        "granttype": "client_credentials"
      },
      "protocol": [
        "amqp10ws"
      ],
      "broker": {
        "type": "sapmgw"
      },
      "uri": "wss://enterprise-messaging-messaging-gateway.cfapps.us10.hana.ondemand.com/protocols/amqp10ws"
    },
    {
      "oa2": {
        "clientid": "sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
        "clientsecret": "0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw=",
        "tokenendpoint": "https://bb8713batrial.authentication.us10.hana.ondemand.com/oauth/token",
        "granttype": "client_credentials"
      },
      "protocol": [
        "mqtt311ws"
      ],
      "broker": {
        "type": "sapmgw"
      },
      "uri": "wss://enterprise-messaging-messaging-gateway.cfapps.us10.hana.ondemand.com/protocols/mqtt311ws"
    },
    {
      "oa2": {
        "clientid": "sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732",
        "clientsecret": "0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw=",
        "tokenendpoint": "https://bb8713batrial.authentication.us10.hana.ondemand.com/oauth/token",
        "granttype": "client_credentials"
      },
      "protocol": [
        "httprest"
      ],
      "broker": {
        "type": "saprestmgw"
      },
      "uri": "https://enterprise-messaging-pubsub.cfapps.us10.hana.ondemand.com"
    }
  ]
}
```

こんなJsonがあったら下記のように変えます。
今回はAMQPプロトコルで連携するのでそこのブロックの情報を持ってきます。


```Javascript
'use strict';

const fs = require('fs');

module.exports = {
  uri: 'wss://enterprise-messaging-messaging-gateway.cfapps.us10.hana.ondemand.com/protocols/amqp10ws',
  oa2: {
    endpoint: 'https://bb8713batrial.authentication.us10.hana.ondemand.com/oauth/token',
    client: 'sb-clone-xbem-service-broker-63fe8d60c2a54ecf99b269e00ccbea62-clone!b40970|xbem-service-broker-!b732',
    secret: '0c51cb83-e3a4-4b60-9814-1a588f5dde40$qNB4NPKpqsxQIzn6TxZf0UgrVob065IWlQjBAxolRYw='
  },
  credentials: {
    mechanism: '',
    user: '',
    password: ''
  },
  data: {
    source: 'queue:SampleForAQueue',
    target: 'topic:sample/for/a/topic',
    payload: new Buffer.allocUnsafe(20),
    maxCount: 100,
    logCount: 10
  }
};
```

[cf-sample-config.js](https://github.com/SAP-samples/event-mesh-client-nodejs-samples/blob/main/xb-msg-amqp-v100-samples/config/cf-sample-config.js)に合わせてEventMeshのqueueとtopicを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/81572425-e880-7841-5931-e116601a1965.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/eca6b2b8-68f0-d153-0824-a5e3b6b09faa.png)


## Queue Subscriptionsの確認
[producer.js](https://github.com/SAP-samples/event-mesh-client-nodejs-samples/blob/main/xb-msg-amqp-v100-samples/src/producer.js)と[consumer.js](https://github.com/SAP-samples/event-mesh-client-nodejs-samples/blob/main/xb-msg-amqp-v100-samples/src/consumer.js)があってproducer.jsはEventMeshに連携を飛ばして、それをconsumer.jsで使う構成になっています。

こちらのコマンドを実行するとproducer実行され100回メッセージが飛ばされます。

```cmd
$ npm run-script producer
> @sap/xb-msg-amqp-v100-samples@1.0.0 producer
> node src/producer.js ../config/cf-sample-config.js

connected node-v12.22.7 @sap/xb-msg-amqp-v100 0.9.56
10
20
30
40
50
60
70
80
90
100
disconnected

messages published    : 100
messages acknowledged : 100
websocket incoming    : 10 frames ( small 5 / medium 5 / large 0 )
websocket outgoing    : 9 frames ( small 6 / medium 3 / large 0 )
protocol incoming     : 108 frames, 3681 bytes, 9 chunks
protocol outgoing     : 107 frames, 10524 bytes, 8 chunks ( reuse 7 / alloc 3 )
drain events incoming : 0
drain events outgoing : 0
socket wait time[ms]  : 0
socket init time[ms]  : 3217
sender work time[ms]  : 19
overall run time[ms]  : 19
overall rate [msg/s]  : 5263
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/2fa29676-b602-1736-0b67-bfeccc5341e8.png)

Queuesが100回分溜まったことが確認できます。
次にconsumerを動かしてQueuesを消費していきます。

```cmd
$ npm run-script consumer
> @sap/xb-msg-amqp-v100-samples@1.0.0 consumer
> node src/consumer.js ../config/cf-sample-config.js

connected node-v12.22.7 @sap/xb-msg-amqp-v100 0.9.56
10
20
30
40
50
60
70
80
90
100
disconnected

messages published    : 100
websocket incoming    : 10 frames ( small 4 / medium 6 / large 0 )
websocket outgoing    : 12 frames ( small 7 / medium 5 / large 0 )
protocol incoming     : 106 frames, 15048 bytes, 10 chunks
protocol outgoing     : 106 frames, 5728 bytes, 11 chunks ( reuse 10 / alloc 3 )
drain events incoming : 0
drain events outgoing : 0
overall rate [msg/s]  : 526
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/9ca47b8c-ef37-3cff-931b-c57a7b3bac06.png)

Queuesが消費されて0になったことが確認できます。
実務の実装ではproducer,consumerの役割を果たすアプリがあってその橋渡しをするのがEventMeshの役割かと思います。

## Webhook  Subscriptionsの確認

EventMeshにWebhook Subscriptionの設定を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/752cbb64-299a-357a-1909-a866956df7ea.png)

Webhookの宛先はこちらを指定しています。

https://webhook.site/#!/3533749f-8663-46aa-a872-49041296edb1

producerをもう一度実行してみます

```cmd
$ npm run-script producer
> @sap/xb-msg-amqp-v100-samples@1.0.0 producer
> node src/producer.js ../config/cf-sample-config.js

connected node-v12.22.7 @sap/xb-msg-amqp-v100 0.9.56
10
20
30
40
50
60
70
80
90
100
disconnected

messages published    : 100
messages acknowledged : 100
websocket incoming    : 10 frames ( small 5 / medium 5 / large 0 )
websocket outgoing    : 9 frames ( small 6 / medium 3 / large 0 )
protocol incoming     : 108 frames, 3681 bytes, 9 chunks
protocol outgoing     : 107 frames, 10524 bytes, 8 chunks ( reuse 7 / alloc 3 )
drain events incoming : 0
drain events outgoing : 0
socket wait time[ms]  : 0
socket init time[ms]  : 3217
sender work time[ms]  : 19
overall run time[ms]  : 19
overall rate [msg/s]  : 5263
```
Queuesは0のままです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/2b41384c-649b-f03d-cad2-3d0fc463de83.png)

webhookの連携先を確認してみます。
2回やったので合計200回のRequestが確認できました。
こうやって溜まったQueuesぶん指定されたURLをコールさせることが出来ます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137987/bd7c7452-dabf-bdba-ef9d-ec2acc51e76b.png)

# 最後に
EventMeshを使うと非同期なシステム間の連携ができることが確認出来ました。
おしまい