Messaging Gateway

Scope#

Equinix Messaging Gateway delivers notifications to an Azure Service Bus queue or subscription that belongs to the customer's EMG subscription. com.eqixiac.equinix.messaging parses those messages into typed records and dispatches them to a handler. The package calls no Equinix REST API.

Notification types#

The parser is tolerant. An EMG envelope of a type the SDK does not model becomes an UnknownNotification, and an enum value the SDK does not list reads as UNKNOWN. EmgParseException is thrown only for a body that is not an EMG envelope.

Parse a message#

import com.eqixiac.equinix.messaging.EmgMessage;
import com.eqixiac.equinix.messaging.EmgMessageParser;
import com.eqixiac.equinix.messaging.IbxNotification;

String json = Files.readString(Path.of("notification.json"));
EmgMessage message = new EmgMessageParser().parse(json);
if (message instanceof IbxNotification n && n.body().isServiceImpacting()) {
    System.out.println(n.id() + " " + n.category() + " " + n.body().startDateTime());
}

Listen on Azure Service Bus#

ServiceBusEmgReceiver needs the optional dependency com.azure:azure-messaging-servicebus 7.18.0, which the SDK does not pass on to its consumers.

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-servicebus</artifactId>
    <version>7.18.0</version>
</dependency>

import com.eqixiac.equinix.messaging.EmgMessageHandler;
import com.eqixiac.equinix.messaging.IbxNotification;
import com.eqixiac.equinix.messaging.transport.EmgListener;
import com.eqixiac.equinix.messaging.transport.EmgReceiver;
import com.eqixiac.equinix.messaging.transport.servicebus.ServiceBusEmgReceiver;

EmgMessageHandler handler = new EmgMessageHandler() {
    @Override
    public void onIbxNotification(IbxNotification n) {
        if (n.body().isServiceImpacting()) {
            System.out.println(n.id() + " " + n.category() + " " + n.body().startDateTime());
        }
    }
};

EmgReceiver receiver = ServiceBusEmgReceiver.forQueue(
    System.getenv("EMG_CONNECTION_STRING"), System.getenv("EMG_QUEUE"));
try (EmgListener listener = new EmgListener(receiver, handler)) {
    listener.run();   // blocks
}

ServiceBusEmgReceiver.forSubscription(connectionString, topic, subscription) reads a topic subscription, and wrap(ServiceBusReceiverClient) takes a client that you built.

The connection string and the queue name come from the Customer Portal: Developer Settings, Messaging Gateway, the subscription, Configuration, Copy YAML. The connection string is a secret.

Listener behaviour#

run() blocks until close() or a thread interrupt. A failed receive call is logged and retried after a back-off. close() waits for the message in flight to be settled, abandons the rest of the current batch so that it is delivered again, and closes the receiver. pollOnce() processes one batch and returns the number of deliveries. The listener logs broker message ids, delivery counts and exception types, and no message body or credential.

Service Bus delivers at least once. A handler must be idempotent: de-duplicate on id().

Other transports#

Implement EmgReceiver to read from another transport. receive(maxMessages, wait) returns EmgDelivery objects, and the listener calls complete(), abandon() or deadLetter(reason, description) on each.

Not implemented#

• EMG Order Management, deprecated by Equinix on 2024-09-30
• client_credentials EMG authentication, deprecated on 2024-03-18
• Signature verification. The header carries signature(), and Equinix documents no verification procedure.
• Attachment download