Wrapper for communication with cap messaging service

Table of Contents

• Features
• Installing
• Configuration
• Implementation
• License

Features

• Simple configuration
• Using placeholders in topics
• Optional logging of whole event message (DEBUG=messaging)
• Set 'x-correlation-id' in event header to follow events over applications
• Managing composite configuration in combination of local or file based messaging
• Fully integrated in cdc runtime without hard binding of services to register for events
• Usage of event prefixes
• Auto configuration with cds plugin feature
• Fixes listener registration, if service name in package.json differs to service name

Deprecated

• Set 'topic' in event header (deprecated: use type and/or source instead)

Installing

Using npm:

$ npm install @amag-ch/cds-messaging

Using yarn:

$ yarn add @amag-ch/cds-messaging

Configuration

Default configuration

{
    "cds": {
        "requires": {
            "queue": {
                "kind": "persistent-outbox"
            },
            "kinds": {
                "composite-messaging": {
                    "impl": "@amag-ch/cds-messaging/composite.js",
                    "format": "cloudevents",
                    "subscribePrefix": "$namespace/",
                    "publishPrefix": "$namespace/"                    
                },
                "enterprise-messaging-http": {
                    "format": "cloudevents",
                    "subscribePrefix": "$namespace/",
                    "publishPrefix": "$namespace/"
                },
                "enterprise-messaging-amqp": {
                    "format": "cloudevents",
                    "subscribePrefix": "$namespace/",
                    "publishPrefix": "$namespace/",                    
                    "amqp": {
                        "incomingSessionWindow": 100
                    }
                }
            }
        }
    }
}

Set or overwrite defaults

Simple default configuration over standard messaging object in package.json

At least only the namespace is needed

{
    "cds": {
        "requires": {
            "messaging": {
                "kind": "enterprise-messaging-http",
                "impl": "@amag-ch/cds-messaging/service.js",
                "credentials": {
                    "namespace": "btp/<org>-<space>/<application>"
                }
            }
        }
    }
}

This results in following standard configuration

{
    "cds": {
        "requires": {
            "messaging": {
                "kind": "enterprise-messaging-http",
                "impl": "@amag-ch/cds-messaging/service.js",
                "subscribePrefix": "$namespace/",
                "publishPrefix": "$namespace/",
                "format": "cloudevents",
                "credentials": {
                    "namespace": "btp/<org>-<space>/<application>"
                }
            }
        }
    }
}

Configure sepcial queues for own application

Simpliest configuration

{
    "cds": {
        "requires": {
            "messaging-self-publishes": {
                "kind": "enterprise-messaging-http"
            }
        }
    }
}

This results in following standard configuration

{
    "cds": {
        "requires": {
            "messaging-self-publishes": {
                "kind": "enterprise-messaging-http",
                "subscribePrefix": "$namespace/",
                "publishPrefix": "$namespace/",
                "format": "cloudevents",
                "credentials": {
                    "namespace": "btp/<org>-<space>/<application>"
                },
                "queue": {
                    "name": "$namespace/self-publishes",
                }
            }
        }
    }
}

Of course, the defaults could be overwritten

{
    "cds": {
        "requires": {
            "messaging-self-publishes": {
                "kind": "enterprise-messaging-http",
                "queue": {
                    "maxDeliveredUnackedMsgsPerFlow": "1"
                }
            }
        }
    }
}

Configure special queues for foreign application events

Simpliest configuration

{
    "cds": {
        "requires": {
            "messaging-foreign-system": {
                "kind": "enterprise-messaging-http",
                "subscribePrefix": "foreign/system/namespace/"
            }
        }
    }
}

This results in following standard configuration

{
    "cds": {
        "requires": {
            "messaging-foreign-system": {
                "kind": "enterprise-messaging-http",
                "subscribePrefix": "foreign/system/namespace/",
                "publishPrefix": "$namespace/",
                "format": "cloudevents",
                "credentials": {
                    "namespace": "btp/<org>-<space>/<application>"
                },
                "queue": {
                    "name": "$namespace/foreign-system"
                }
            }
        }
    }
}

Configure messaging with multiple queues with composite procedure

{
    "cds": {
        "requires": {
            "messaging": {
                "kind": "composite-messaging",
                "default": "messaging-default",
                "credentials": {
                    "namespace": "btp/<org>-<space>/<application>"
                },
                "routes": {
                    "messaging-foreign-system": [
                        "foreign-system:object/changed",
                        "other/object/changed"
                    ]
                }
            },
            "messaging-default": {
                "kind": "enterprise-messaging-amqp",
                "[development]": {
                    "kind": "file-based-messaging"
                }
            },
            "messaging-foreign-system": {
                "kind": "enterprise-messaging-http",
                "subscribePrefix": "foreign/system/namespace/"
            }
        }
    }
}

• A default service definition is required (property default). The name of the default service doesn't matter, but it shouldn't be a service for external events
• Events are everytime emitted through the default service, except those routes to a local-messaging definition
• Events in on handlers, which are not defined in the routes, are handled by the default service
• If the default service is not real messaging service (e.g local or file-based), then only the default service would be created and handle all. Subscribe prefixes are still been considered.
• Events could be prefixed with pattern [a-zA-Z_-]: to avoid naming collisions with own events in configuration

Implementation

Own application events

const messaging = await cds.connect.to('messaging')
const { buildEventFromTemplate } = require('@amag-ch/cds-messaging')

cds.once('listening', () => {

    /*
    * Standard emitting
    * Signature is same as for standard cap messaging service
    * 
    * Full topic based on configuration is: $namespace/object/changed => cap/<space>/<application>/object/changed
    */
    messaging.emit('object/changed', { ID })  
    
    /*
    * Emitting with replacements
    * Usefull, if cannot fully control the input value  
    * 
    * Following call results in event: objectwithillegalcharacters/changed
    */    
    messaging.emit({ template: ':1/changed', replacements: ['object-with-illegal_characters']}, { ID })

    /*
    * Emitting with replacements
    * Usefull, if cannot fully control the input value  
    * 
    * Following call results in event: objectwithillegalcharacters/changed
    */    
    messaging.emit(buildEventFromTemplate(':1/changed', ['object-with-illegal_characters']), { ID })    
})

/**
 * Topics, we are listening here are (based on subscribePrefix):
 * - $namespace/object/changed => cap/<space>/<application>/object/changed
 * - $namespace/objectwithillegalcharacters/changed => cap/<space>/<application>/objectwithillegalcharacters/changed
 */
messaging.on([
    'object/changed',
    'objectwithillegalcharacters/changed'
], async (msg) => {

    console.log(msg)
    ...
})

Foreign application events

const messaging = await cds.connect.to('messaging')

/**
 * Topic, we are listening here is (based on subscribePrefix): foreign/system/namespace/object/changed
 */
messaging.on('foreign-system:object/changed', async (msg) => {

    console.log(msg)
    ...

    /**
     * This results in topic (based on publishPrefix): $namespace/object/changed => cap/<space>/<application>/object/changed
     */
    messaging.emit('object/changed', { ID })  
})

Integrated in service implementations

/**
 * Prefixes needed, to register to other unknown services
 */
this.on([
    'foreign-system:object/changed',
    'self:objectwithillegalcharacters/changed'
], async (msg) => {

    console.log(msg)
    ...

    /**
     * Event definition in cds service needed
     * 
     * @topic: 'object/changed'
     * event ![object/changed] { ID: UUID }
     */
    this.emit('object/changed', { ID })

    /**
     * Event definition in cds service needed
     * 
     * @topic: 'object/created'
     * event objectCreated { ID: UUID }
     */
    this.emit('objectCreated', { ID })     
})

• Annotation @topic should be used to avoid prepending of service name to the resulting event
• Usage of template based events is not possible here

License

MIT