CAP – Specifying the queues to be consumed from SAP Event Mesh

Hi,

We are trying this and have a few issues. First, the queuing service is only connecting if we keep the exact name "messaging" in the configuration.

... this works
    "cds": {
        "requires": {
            "messaging": {
                "kind": "enterprise-messaging-shared",
                "queue": {
                    "name": "default/my/queue",
                    "maxDeliveredUnackedMsgsPerFlow": 5,
                    "maxRedeliveryCount": 10,
                    "deadMsgQueue": "default/my/dmq",
                    "maxTtl": 0,
                    "respectTtl": true
                }
            }
... but just changing "messaging" to "messaging1" makes the queue not connecte
            "messaging1": {
                "kind": "enterprise-messaging-shared",
                "queue": {
                    "name": "default/my/queue",
                    "maxDeliveredUnackedMsgsPerFlow": 5,
                    "maxRedeliveryCount": 10,
                    "deadMsgQueue": "default/my/dmq",
                    "maxTtl": 0,
                    "respectTtl": true
                }
            }

The service implementation is like this:

module.exports = srv => {
    cds.connect.to("messaging").then(messaging => {
        messaging.on("default/my/queue", async (msg) => {
            const messagePayload = JSON.stringify(msg)
            console.log('===> Received message : ' + messagePayload)
        });
    });
}

We also notice that the dead message queue does not get created. A cosmetic issue (if we create it first it gets properly connected). But with that sorted we are still not getting messages to pass over to the dead message queue. We are throwing a simple error from the handler function and the result of that is that the message keeps getting repeated (but never goes into the dmq).

Do you have any input on what we are doing wrong?

Thanks in advance!

//Carl

Hi,

Thanks for the quick reply. This is what we have been trying, sorry I was not clear on that aspect in the first comment. With "messaging" in "cds.requires" and in "conect.to" we can see in the output that it connects, like this:

[cds] - server listening on { url: 'http://localhost:4004' }
[cds] - launched in: 6.741s
[ terminate with ^C ]

[messaging] - Create messaging artifacts 
[messaging] - Create queue { queue: 'default/my/queue' }
[messaging] - Get subscriptions { queue: 'default/my/queuee' }
[messaging] - Unchanged subscriptions [ 'default/my/queue' ] 

If we have "messaging1" in both places, the log stops after "terminate with ^C" and the messaging does not work.

I tried to isolate it a bit further (on version 5.4.3). I think it is related to this.

// in @sap/cds/server.js, line 47, there is a dedicated connect to 'messaging' if specified

  // connect to essential framework services if required
  const _init = o.in_memory && (db => cds.deploy(csn).to(db,o))
  if (cds.requires.db) cds.db =  await cds.connect.to ('db') .then (_init)
  if (cds.requires.messaging)    await cds.connect.to ('messaging')             //<<<<<<<<<<<<<<<<<<<<<<
  if (cds.requires.multitenancy) await cds.mtx.in (app)

  // serve all services declared in models
  await cds.serve (o.service,o).in (app)
  await cds.emit ('served', cds.services)               //> hook for listeners

// in @sap/cds/libx/_runtime/messaging/AMQPWebhookMessaging.js, line 17 there is a "startListening" hook which does the connection

    cds.once('listening', () => {
      this.startListening()
    })

The "connect.to("messaging")" in "server.js" causes the "subscribedTopics" of the AMQPWebhookMessaging to be populated before the "listening" event occurs. If we use a different name they occur in the wrong order (subscribedTopics is not populated when the startListening method is called and nothing is connected).

Adding this to a custom server.js makes the other-named messaging instance connect early and then it works properly.

const cds = require('@sap/cds')
module.exports = async (o) => {
    // Workaround to connect messaging services with a different name than "messaging"
    await cds.connect.to("messaging1");
    // Delegate to default server
    return cds.server(o);
}

Hope this helps finding the root cause.

Thanks!

//Carl

Perfect, thanks! That did the trick!