Send AMQP messages from CPI to Enterprise Messaging and Consume them in a Node.js AMQP Application ?

[vvdries]

Hi Cloud Platform Integrator!

In this blog post we will have a look on how to send messages from CPI via AMQP to the Enterprise Messaging Service. Once the messages arrive in the Queue, it will be consumed in a Node.js application running on Cloud Foundry.

This is a Blog series with the following topics in separated Blog posts:

Cloud Foundry Enterprise Messaging Webhooks ?

Send AMQP messages from CPI to Enterprise Messaging and Consume them in a Node.js AMQP Application ? (this blog)

WebSocket’s in SAP UI5 Multi Target Applications consuming Enterprise Messaging ?

Introduction

To not reinvent the wheel and to credit earlier blogposts of fellow Community Members, I would like to point out the following blog posts:

Cloud Integration – Connecting to Messaging Systems using the AMQP Adapter

https://blogs.sap.com/2019/11/20/cloud-integration-connecting-to-external-messaging-systems-using-th...

Cloud Integration – How to Connect to an On-Premise AMQP server via Cloud Connector

https://blogs.sap.com/2020/01/17/cloud-integration-how-to-connect-to-an-on-premise-amqp-server-via-c...

SAP CPI – AMQP Channel with Cloud Event Broker – SOLACE

https://blogs.sap.com/2020/02/08/sap-cpi-amqp-channel-with-cloud-messaging-solace/

This because these blogposts helped me to get started with AMQP Messaging Channels in Cloud Platform Integration. They are very detailed, well explained and worth to have look at.

In this blog post we will continue working on our earlier configurations, setup and developments from the previous blog post “Cloud Foundry Enterprise Messaging Webhooks ?”.

The prerequisites for this blog are:

1. A configured Queue called “ErrorQueue” in the Enterprise Messaging Cockpit.

2. Stop the earlier created Node.js application in the SAP CF Space, by pressing the stop-icon.

Once this application has stopped, you will see the “Requested State” has switched to “Stopped”.

3. Remove the current and earlier created Webhook Subscription in the Enterprise Messaging Cockpit.

This because a new Node.js App will be created which will consume these AMQP messages from the “ErrorQueue”. If the Webhook would still be subscribed, it would consume the messages from this Queue and there would be no more messages to be consumed by the new Node.js AMQP app.

 

Configuring the Cloud Platform Integration Flow

Since these blog posts are about demoing the capabilities off all these services in a combined End-To-End solution, I will briefly present this CPI-Flow.

This flow will only run once when it is deployed, because it is configured int the “Start Timer” control. This flow continues with the “Trigger Error Script” which holds the following code:

import com.sap.gateway.ip.core.customdev.util.Message;

import java.util.HashMap;

def Message processData(Message message) {

    try {

        def result = 3 * "7";

    } catch (Exception e) {

        throw e;

    }

    return message;

}

This code will try to multiply 3 with the string 7, which will fail and will throw an exception. This will call the “Exception Subprocess 1” that is going to catch this thrown error.

This “Exception Subprocess 1” holds a “Groovy Script” control with the following content:

import com.sap.gateway.ip.core.customdev.util.Message;

import java.util.HashMap;

def Message processData(Message message) {

    byte[] bytes = '''{"errorMessage": "Error Message From CPI Flow!"}''';

    message.setBody(bytes);

    return message;

}

It will create a byte’s array and will set the bytes as body for the message to be send.

When this JSON payload message is not converted to a byte’s array, inconsistency with encoding will occur in the consuming Node.js app.

The outgoing connection to the “Receiver” its “Adapter Type” is “AMQP” while the “Transport Protocol” is of the type “WebSocket”.

Configuration parameter	Value
Adapter Type	AMQP
Transport Protocol	WebSocket

The AMQP “Connection” tab requires the following configuration:

Configuration parameter	Value
Host	enterprise-messaging-messaging-gateway.cfapps.eu10.hana.ondemand.com
Port	443
Path	/protocols/amqp10ws
Proxy Type	Internet
Connect with TLS	Checked X
Authentication	OAuth2 Client Credentials
Credential Name	{your-credential-name}

The “Host” and “Path” will be the same for every implementation. This because it has been setup like this in the Enterprise Messaging Service by SAP. This information can be found in the “Cloud Foundry Space > Service instances > {Your-EM-Service} > Service Keys > {Your-EMService-Key}“.

The AMQP “Processing” tab requires the following configuration:

Configuration parameter	Value
Destination Type	Queue
Destination Name	queue:ErrorQueue
Expiration Period (in s)	(optional)
Delivery	Persistent

More information about sending and receiving on Queues and Topics (Destination Name) can be found here:

https://help.sap.com/viewer/bf82e6b26456494cbdd197057c09979f/Cloud/en-US/72ac1fad2dd34c4886d672e66b2...

The last step is providing the OAuth2 Client Credentials in the CPI Environment under the “Security Material” as new “OAuth2 Credentials”.

Configuration parameter	Value
Name	Provided name in the CPI Flow as Credential name.
Grant Type	Client Credentials
Description	(optional)
Token Service URL	The “tokenendpoint” property in the Enterprise Messaging Service Key configuration under the AMQP section.
Client ID	The“clientid” property in the Enterprise Messaging Service Key configuration under the AMQP section.
Client Secret	The“clientsecret” property in the Enterprise Messaging Service Key configuration under the AMQP section.
Client Authentication	Send as Request Header
Include Scope	Unchecked

 

Deploying the Cloud Platform Integration Flow

This was a brief explanation of the CPI-Flow. Deploy this flow and see the number of messages inside the “ErrorQueue” increasing with 1 every time you deploy the flow. The flow will fail and the “Exception Subprocess 1” will created the message inside the “ErrorQueue” every time the flow fails.

 

Consume the AMQP Message in a Node.js App

A new Node.js application will be created, to setup a WebSocket that listens for AMQP messages on the “ErrorQueue” Queue.

The SAP HANA Academy already build such and application and shared it in a repository on GitHub. Clone this repository to your local IDE:

git clone https://github.com/saphanaacademy/em-consumer.git

The GitHub repository of this application from the SAP HANA Academy can be found here:

https://github.com/saphanaacademy/em-consumer

Inside the “mta.yaml” file change the green marked values with the name of your “Enterprise Messaging Service Instance” and the blue marked ones with “queue:ErrorQueue”. This because this is the Queue where the messages from CPI are subscribed to.

The next changes take place inside the “server.js” file. Change the following piece of code:

The code looks like this:

stream

  .on('data', (message) => {

    const payload = JSON.parse(message.payload.toString());

    console.log('Error: ' + payload.errorMessage);

    message.done();  

});

The encoding is changed in the “.toString()” function and the value is logged to the console starting with “Event:”.

Remove the obsolete code in the file, since we do not require it for the moment.

The “server.js” file should look like this:

const express = require('express');

const app = express();



var cfenv = require('cfenv');

var appEnv = cfenv.getAppEnv();

var emCreds = appEnv.getServiceCreds(process.env.EM_SERVICE);

var emCredsM = emCreds.messaging.filter(function (em) {

  return em.protocol == 'amqp10ws'

});



const options = {

  uri: emCredsM[0].uri,

  oa2: {

    endpoint: emCredsM[0].oa2.tokenendpoint,

    client: emCredsM[0].oa2.clientid,

    secret: emCredsM[0].oa2.clientsecret

  },

  data: {

    source: process.env.EM_SOURCE,

    payload: new Buffer.allocUnsafe(20),

    maxCount: 100,

    logCount: 10

  }

};



const {

  Client

} = require('@sap/xb-msg-amqp-v100');

const client = new Client(options);

const stream = client.receiver('ErrorQueue').attach(options.data.source);



stream

  .on('data', (message) => {

    const payload = JSON.parse(message.payload.toString());

    console.log('Error: ' + payload.errorMessage);

    message.done();

  });



client

  .on('connected', (destination, peerInfo) => {

    console.log('Connected!');

  })

  .on('assert', (error) => {

    console.log(error.message);

  })

  .on('error', (error) => {

    console.log(error);

  })

  .on('disconnected', (hadError, byBroker, statistics) => {

    console.log('Disconnected!');

  });



client.connect();





const port = process.env.PORT || 3000;

app.listen(port, function () {

  console.info("Listening on port: " + port);

});

Use the following command to push the application to the Cloud Foundry Space:

cf push {optional-name-of-your-app}

Redeploy the Cloud Platform Integration Flow and see the error message from the CPI-Flow appear in the logs of the Node.js application in the Cloud Foundry Space:

Resend a message via Postman to the Queue and the message will also be consumed by the app:

 

Wrap up ?

This blog described how to send messages via the Cloud Platform Integration AMQP Adapter with the WebSocket Transport Protocol. The error messages from the CPI-Flow arrive in the dedicated Queue and are consumed by a Node.js Application running on Cloud Foundry.

In the next blog "WebSocket’s in SAP UI5 Multi Target Applications consuming Enterprise Messaging ?" WebSockets in SAPUI5 and Node.js will be explained, along with their usage and how to consume Enterprise Messaging messages in UI5 applications in real-time.

 

See you in the next blog!

Kind regards,

Dries