AMQP Connectivity to the ASX


What's needed / provided?

Connectivity to the ASX is operating system and programming language agnostic and can be established through a standard TCP/IP socket. The broker is setup to listen on a specific port for incoming connections and leverages TLS certificates for authorisation. 

Users can choose which AMQP 1.0 compliant client they use. The following link provides a number of available AMQP clients. The ASX demonstration client shown in the AMQP working group was developed with Apache Qpid.

ASX used Qpid library for the demonstration client the following reasons:

  • it is open source and subsequently free;
  • it supports many languages (Python, C++, Java, etc.);
  • it has open support forums; and
  • it is well documented with API, tutorials and examples.

User Requirements

  • An AMQP v1.0 compliant client – Apache Qpid or ActiveMQ for example.
  • All connections to the AMQP broker and queues will be via ASX Net (includes IPSec VPN option).
  • A properly configured user-side client capable of submitting messages to the ASX; configurations include TLS certificate, queue name, and connection end-point. Certificates, queue names, and end-point names will be provided by the ASX.
  • A system capable of ISO 20022 XML compliant message creation. Users will be able to download message XSDs from the SWIFT MyStandards website, this will provide a machine readable format.

Client Settings

Most AMQP v1.0 client software requires configuration as follows:

  • Connection Settings - the URL, Port and mutual TLS configuration,
  • Reconnection Settings - retry-configuration in case the connection to ASX is disrupted,
  • Session Settings - an AMQP concept, these are different for sending or receiving messages,
  • Message Producer or Message Consumer Settings - again, different when you are sending or receiving messages, and
  • Message Settings - mainly important when you are sending messages.


Disclaimer

Most of the technical details, below, describes the QPID AMQP 1.0 Client, which presents a "JMS" (Java Message Service) API to the calling software.  This Client is also available for C++, C# and Python2.

Despite the mainly-java descriptions, the actual Settings are much the same for all invocations of the QPID client and those proprietary clients based on the QPID client.

Preparation

Before you can connect to an ASX test or production AMQP messaging service, you must register on ASX' Customer Service Account Management (CSAM) Portal, part of ASX Online, for that environment.  This is used to register your access requirements and request your client TLS certificate for authentication to ASX.  Note that ASX' AMQP Server's certificate is signed by DigiCert so your operating system or AMQP execution-environment should already have the correct DigiCert root Certificate Authority certificate to trust the AMQP server.  (You may wish to download the ASX-signed Certificate-Authority ("CA") certificates so you can verify your downloaded TLS certificate issued by the ASX and to trust the ISO 20022 Signing certificate used by ASX to sign egress messages.)  You will also be advised of the AMQP URL, Port and queues you can access and further instructions.

Refer to Section 06 - CSAM Security Details for more details.

Connection Settings

Once you have registered to access a test or production environment via the CSAM portal, you will need to configure the connection properties for your AMQP Client.

There is a large number of AMQP 1.0 Clients but many are based on the QPID client so the property names used here are for that client.  Refer to the documentation of your particular AMQP 1.0 client to determine the property names and how to set them.  Some require a URL-format connection-string, where the properties are query-parameters, some require configuration-files while others have property-panes.

PropertySample ValueDescription
host *ite1-amqp.asx.com.auFully-Qualified domain name of the ASX AMQP service
port *4005Port assigned to you for that environment

tcpKeepAlive

TRUE

Do not drop the connection if idle.  There is a considerable performance overhead to re-establish the connection.

contextProtocol

TLSTransport-Layer Security - This is required by ASX for encryption of AMQP traffic between your Client and the ASX AMQP server.  In addition, ASX requires you to present a Client Certificate, which will be verified by ASX and used to authorise access to your queues.
TLS Trust-Store Settings:

The trust-store must contain the ASX Certificate-Authority ("CA") certificates you downloaded from the CSAM portal, so you can trust the ASX AMQP server.

Note that you may need to put this downloaded "CA-Cert" in a separate truststore and reference it here, add it to your operating-system or VM certificate store, or you may already have the root and intermediate CA Certificates.  Depending on your client and operating system, you may or may not need to explicitly reference this truststore in your connection settings.

Refer to the documentation of your particular AMQP Client to determine how to configure TLS to trust the ASX AMQP server

trustStoreLocation

truststore.pkcs12

Define the file-path to your trust-store, if your AMQP Client requires it for TLS, rather than automatically refers to the operating-system or VM's certificate store.

trustStorePassword **

notverysecret

Define the password to your trust-store if necessary.

trustStoreType

pkcs12

Define the type of trust-store file, if necessary.
TLS Key-Store Settings:

The ASX AMQP server will ask your client for your Client TLS Certificate, downloaded from the CSAM Portal, so you may be authenticated and authorised to access your queues.

Note that you may need to put this downloaded "Client-Certificate" in a separate keystore and reference it here, or add it to your operating-system or VM certificate store or "wallet".  Depending on your client and operating system, you may or may not need to explicitly reference this keystore in your connection settings.

Refer to the documentation of your particular AMQP Client to determine how to configure TLS to present your Client-Certificate to ASX for mutual-TLS authentication.

keyStoreLocation

asx_amqpclient_keystore.pkcs12

Define the file-path to your keystore, if your AMQP Client requires it for TLS, rather than automatically refers to the operating-system or VM's certificate store.

keyStorePassword **

secret123

Define the password to your keystore if necessary.

keyStoreType

pkcs12

Define the type of keystore file, if necessary.

Connection Notes

*  You will be advised of the specific host and port when you register to access an ASX AMQP environment

** Generate your own truststore and keystore passwords if you need to reference them in your connection settings

Note that these settings are the same whether you intend to send or receive messages over AMQP.  It is recommended to share the same connection with all of your AMQP sessions, message-producers and message-consumers as there is a considerable performance overhead in establishing a connection.

Reconnection Settings

To connect to AMQP, you will be provided with a single URL/Port per environment.  Most failures of ASX AMQP infrastructure will be handled transparently within ASX but the AMQP Client may experience a momentary loss of connection.  If your AMQP Client supports reconnection parameters, their use will help your AMQP Client survive network and server glitches transparently, without you having to write special code.

The most common AMQP Client, QPID, offers reconnection parameters as part of its “Failover” facility.  Normally, if a server fails, the “failover:” protocol attempts to reconnect to the next server in a list but any actual failover is handled internally by ASX.  However, the reconnection properties still work and handle network glitches for you automatically, even with only the single ASX AMQP URL/Port in the failover "list".

PropertyRequired Value*Description

initialReconnectDelay (ms)

10000

Wait 10 seconds before reconnecting

reconnectDelay (ms)

10000

Subsequent retry every 10 seconds,

useReconnectBackOff

TRUE

…Increase wait time after each attempt

reconnectBackOffMultiplier

2.0

…by 2.  ie. double wait time after each attempt

maxReconnectDelay (ms)

30000

…until wait time is 30 seconds, then just retry every 30secs

maxReconnectAttempts

-1

or

20

Keep retrying forever

or

Specify a maximum number if you need to gain control from the client periodically between reconnection attempts, or if you really want to terminate the client.

startupMaxReconnectAttempts

10

Only try 10 times if connection has never been successful

warnAfterReconnectAttempts

10

Only log warning-message every 10 attempts

randomize

FALSE

Not applicable to AMQP at ASX

amqpOpenServerListAction

IGNORE

ASX Server might suggest other hosts or ports. Ignore them

* Accreditation

The reconnection-intervals will form part of the accreditation and must not be any more-frequent than 3 seconds.

Refer to Connectivity Accreditation Overview for more details.

Sample QPID Connection URL

Sample QPID AMQP 1.0 connection URL with Connection and Reconnection (Failover) options:

failover:(amqps://ite1-amqp.asx.com.au:4005?
transport.tcpKeepAlive=true&transport.contextProtocol=TLS&
transport.trustStoreLocation=truststore.pkcs12&transport.trustStorePassword=notverysecret&
transport.keyStoreLocation=asx_amqpclient_keystore.pkcs12&transport.keyStorePassword=secret123)?
failover.initialReconnectDelay=3000&failover.reconnectDelay=3000&
failover.maxReconnectDelay=30000&failover.useReconnectBackOff=true&
failover.reconnectBackOffMultiplier=2.0&failover.maxReconnectAttempts=-1&
failover.startupMaxReconnectAttempts=10&failover.warnAfterReconnectAttempts=10&
failover.randomize=false&failover.amqpOpenServerListAction=IGNORE

Settings for Sending Request Messages

Session Settings

Create your AMQP Sessions just after you create your connection when you initialise your Client, or if you want to increase the number of threads to increase throughput.  You can then create a Message Producer to start sending messages to a queue.  If you have applied for multiple sets of queues by special arrangement, ASX recommends creating a separate session for each queue as the queues are not related.

Transactional Sessions VS Non-Transactional

ASX recommends Transactional settings for sessions to be used for sending Request messages due to slight performance improvements and greater control within your AMQP client software.  However, if your message volumes are not great, non-transactional sessions may suffice.

  • Transactional may provide better performance when sending large numbers of request-messages between "commits".  Although AMQP only supports "local" transactions (ie. which can only commit messages sent via the AMQP-Session, not coordinated with, say, a database).  The Session should be committed immediately before committing the source of the messages, either another Queue or Database.  The client may cache messages locally and send them to the AMQP Server only when the cache is full or when the session is committed, reducing network overheads and improving throughput.
  • Non-Transactional sessions can allow "auto-acknowledgement", where the AMQP server will return an acknowledgement to the client immediately after the message has been persisted securely.  Although simpler, the extra network turn-arounds might reduce throughput slightly.

Note that "notify" messages from ASX are not sent immediately in response to any request messages you send so there is no need to wait for a notify-message in response.  Since the requests and notifies are unrelated, it is best to have separate sessions for sending and receiving anyway, apart from the sending and receiving having different recommended session-settings.

PropertyTransactional ValueNon-Transactional ValueDescription
isTransactedtruefalseIndicates if session to be transactional or not.
acknowledgeMode0
(SESSION_TRANSACTED)
1
(AUTO_ACKNOWLEDGE)

If transacted, you must "commit" the session after you've sent some messages.  If the commit fails, you must resend all of those messages since the last commit or since the session was created.

If not transacted, AUTO_ACKNOWLEDGE mode means that the call to the Message Producer's "send" method will either return successfully, indicating your message was stored at ASX successfully, or will throw an exception indicating a failure.

Reconnection

If the SEND failure was caused by a network or ASX infrastructure problem, using the recommended Reconnection Settings will handle most problems transparently, only returning the exception if it could not be handled internally.  In this case, your Client software must resend:

  • Transactional: all messages since the last commit,
  • Non-Transactional: just the last message.

Message Producer Settings

Once you have created a Session, you can then create a Message Producer for the request queue.  The queue name is the only option.  The Message Producer can be used again and again to send messages to that queue. 

PropertySample ValueDescription
destination"sharesrus.00001.ite1.posttrade.csp.amqp.trx.request"This is either the request queue name provided after you register via the CSAM portal, or some other destination-definition object representing that queue.

Message Settings

AMQP messages have a Body (“Text” type for ASX), Properties (name/value pairs, like HTTP Headers) & Attributes (Expiry, Persisted, etc)

Message PropertiesNameSample ValueDescription
Bodyn/an/aThe message-body must be a "Text" type and must be signed ISO 20022 XML.  There are instructions for signing your request message under ISO Message Signing
Properties"BizMsgIdr""00001|22222336600"

This is the <BizMsgIdr> value from the request-message's Business Application Header (BAH)

Setting this message "Property" allows ASX Customer Technical Support staff to help track and trace messages you send and advise whether your message has been received, should there be any doubt.  It avoids the need to parse the BizMsg body to find this field.

Note that any messages sent to you on your "trx.notify" queue will have new BizMsgIdr values, not this one from the request.

Special note for AMQPLite users

AMQPLite users should populate the property MessageId with the BizMsgIdr value, this is due to limitation AMQPLite has a fixed list of properties.


"correlationId""00001-22222336600"

This is also the <BizMsgIdr> value from the request-message's BAH but with the vertical-bar character replaced by a hyphen.

The correlationId will be received and stored by CHESS and set in any resulting "trx.notify" messages that relate to this request.

Important

A system limitation prevents the vertical-bar character ("|") being saved and returned back to you so you must use a hyphen ("-")  when setting this "correlationId" property.

Attributesn/an/a

There is no need to set any message attributes when sending it using the Message Producer.

ASX requires that all messages are be persisted and non-expiring (ie. Not lost if there is an outage) and the AMQP Servers have been set up with this as default behaviour.

Important

No attempt must be made to set a message-expiry-time (or "time to live") or to alter the the message to non-persistent (ie. "not saved to disk").  Message loss may result.

You may set Message Priority if you want some messages to be received by ASX ahead of other messages in your queue.  Priority 0 is the lowest priority while Priority 9 is the highest priority.  The default is 2.

Once you have created the Message object as above, you simply invoke the Message Producer's "Send" method to send the message to the request queue associated with that producer.

Sample Pseudo-Code

The following sample code is to assist developers.

For production use, all AMQP clients must maintain continuous connections and sessions unless disconnected by either party. In the sample below, connections and sessions are disconnected when there are no more messages to process. This design pattern is purely for example purposes.

Send Request messages
try {
    ConnectionFactory factory = 
        new JmsConnectionFactory("failover:(amqps://ite1-amqp.asx.com.au:4005?contextProtocol=TLS&keyStorePath=.....)?failover.initialReconnectDelay=3000.....");
    Connection connection = factory.createConnection();
    connection.setExceptionListener(new MyExceptionListener());
    connection.start();
    
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);    
    MessageProducer messageProducer = session.createProducer(new JmsQueue("sharesrus.00001.ite1.posttrade.csp.amqp.trx.request"));
    
    SignedIso20022Message myIsoMsg = getNextSignedIsoMsgRequests();                           // GET FIRST ISO REQUEST MSGS FROM SOMEWHERE
    while (isoMsg != null) {
        TextMessage message = session.createTextMessage(myIsoMsg.getXmlText());
        message.setStringProperty("BizMsgIdr",     myIsoMsg.getBizMsgIdr());                  // GET BIZ-MSG-ID OF ISO20022 REQUEST
        message.setStringProperty("correlationId", myIsoMsg.getBizMsgIdr().replace('|','-')); // SAME BIZ-MSG-ID, BUT WITH HYPHEN
        messageProducer.send(message);
        LOG.info("Sent id=" + message.getJMSMessageID() + ",body=" + myIsoMsg.getXmlText());
        myIsoMsg = getNextSignedIsoMsgRequests();                                             // GET NEXT ISO REQUEST MSGS FROM SOMEWHERE
    }
    session.close();
    connection.close();
} catch (Exception exp) {
    LOG.error(exp.toString());
    System.exit(1);
}

Settings for Receiving Notify Messages

Session Settings

Create your AMQP Sessions just after you create your connection when you initialise your Client, or if you want to increase the number of threads to increase throughput.  You can then create a Message Consumer to receive messages from a queue.  If you have requested the separate "Report" notify queue or applied for multiple sets of queues by special arrangement, ASX recommends creating a separate session for each queue as the queues are not related.

Transactional Sessions VS Non-Transactional

ASX recommends Transactional settings for sessions to be used for receiving Notify messages if you want to handle errors and not terminate the session or connection.  Transactions allow you to explicitly "rollback" the session so the notifications can be received again.

If your error-handling is fairly simple and you just let all exceptions terminate the process and the Connection, you may as well just use non-transactional sessions and acknowledge your messages if processing was successful.  You cannot explicitly "rollback" anything without closing the entire Connection.

  • Transactional sessions when consuming messages off a queue provide the ability to explicitly "roll-back" messages received in the last transaction so they can be received again.  This is necessary if your client error-handling needs to "replay" messages if a temporary error occurred.  (Note: if there was something wrong with the message, you should log it and contact ASX but not roll-back the transaction, otherwise you will received this "poisoned message" again and again.)

    AMQP only supports "local" transactions (ie. which can only commit messages received via the AMQP-Session, not coordinated with, say, a database).  You just have to remember to "Commit" your session just after you commit your database updates from processing the received messages.
  • Non-Transactional sessions require the Client to "Acknowledge" receipt of a message from ASX.  ASX requires clients to explicitly Acknowledge messages only after they are saved or processed successfully, so they may be resent if your process fails and terminates the connection without acknowledging the received messages.

    You can even save or process a number of messages you received and then "Acknowledge" the last one, which acknowledges all intermediate messages. Using this feature, after you save or process your series of messages, you can commit your database updates, then acknowledge your AMQP so ASX knows not to resend those messages.  This is similar behaviour to the local transactionality AMQP offers anyway but, as mentioned above, you cannot explicitly roll-back the messages onto the queue; you need to terminate your entire process, breaking the Connection to AMQP.

Duplicate Messages

You must expect some messages to be received multiple times, either because your process failed and ASX has resent un-acknowledged notify messages or some upstream process has detected a problem or timed-out and sent the message again.

You must discard any notify-message you receive if you have already successfully processed that BizMsgIdr.

Note that "notify" messages from ASX are not sent immediately in response to any request messages you send.  Since the requests and notifies are unrelated, it is best to use separate sessions for sending and receiving anyway, apart from the sending and receiving having different recommended session-settings.

PropertyTransactional ValueNon-Transactional ValueDescription
isTransactedtruefalseIndicates if session to be transactional or not.
acknowledgeMode0
(SESSION_TRANSACTED)
2
(CLIENT_ACKNOWLEDGE)

If transacted, you must "commit" the AMQP session after you've successfully saved or processed the received notify-messages and committed those updates in your database or system.

If not transacted, CLIENT_ACKNOWLEDGE mode means that you must call the "acknowledge" message-method after you've successfully saved or processed the received notify-messages and committed those updates in your database or system, same as for transacted sessions.

In either case, if processing the received message encounters a problem, if you throw an exception and recreate another session, ASX will resend any unacknowledged or uncommitted messages.

Message Consumer Settings

Once you have created a Session, you can then create a Message Consumer for the notify queue.  The queue name is the only option.  The Message Consumer is used again and again to receive messages from that queue. 

PropertySample ValueDescription
destination"sharesrus.00001.ite1.posttrade.csp.amqp.trx.notify"This is either the Transaction-Notify or Report-Notify queue name provided after you register via the CSAM portal, or some other destination-definition object representing that queue name.

Message Receive Settings

Once you have created the Message Consumer, you can simply invoke the Consumer's "receive" method to receive a message if there are notify messages in the queue.  You will need to invoke this within a loop to retrieve messages.  If there are no messages, specifying a wait time will allow the client to wait for a while for a message to arrive, reducing network traffic if you just kept trying with no wait time.  If there are still no messages after the wait-time, it will return null.  You would normally only terminate this loop when you want to shut down your AMQP message consumer such as at end-of-day, or if there is an unrecoverable problem when you try to save or process the received notify messages.

PropertySample ValueDescription
timeout (ms)30000If there are no messages on the queue, the "receive" method will wait for the specified time for a message to arrive.  If there are still no messages, it will return null.

Sample Pseudo-Code

The following sample code is to assist developers.

For production use all AMQP clients must maintain continuous connections and sessions unless disconnected by either party, in the sample below connections and sessions are established when there are no more messages to process, this design pattern is purely for example purposes.

Receive Notify messages
try {
    ConnectionFactory factory = 
        new JmsConnectionFactory(“failover:(amqps://ite1-amqp.asx.com.au:4005?sslEnabled=true&keyStorePath=.....");
    Connection connection = factory.createConnection();
    connection.setExceptionListener(new MyExceptionListener());
    connection.start();
    
    Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);    
    MessageConsumer messageConsumer = session.createConsumer(new JmsQueue(“sharesrus.54321.ite1.posttr……trx.notify”));

    while(checkIfConsumptionToContinue()) {                              // CUSTOM CODE TO DECIDE CONTINUE VS STOP-NOW
        Object msgObj = messageConsumer.receive(30000L);                 // WAIT 30 SECONDS IF NO NOTIFICATIONS
        if (msgObj != null && msgObj instanceof TextMessage) {
            TextMessage message = (TextMessage) msgObj;
            LOG.info(“Received correlationId=" + message.getStringProperty(“correlationId”) 
                    + “BizMsgIdr=“ + message.getStringProperty(“BizMsgIdr”) + ",body=" + message.getText());        
            processMsg(message);                                         // OMIT DUPLICATES, CHECK SIGNATURE, PROCESS/SAVE
            message.acknowledge();                                       // ACKNOWLEDGE REMOVES MSG FROM QUEUE
        } // TextMessage
    }

    session.close();
    connection.close();
} catch (Exception exp) {
    LOG.error(exp.toString());
    System.exit(1);
}

Important

Remember to:

  • Check for notify-messages with duplicate BizMsgIdr values!  Some messages will be resent under some error or timeout conditions.
  • Verify the received message against the provided Signature and advise ASX if there is a discrepancy.
  • Acknowledge your message(s) or Commit your session AFTER you have saved or processed the received messages and committed the updates in your system.



Related Pages:

There are no related labels.

Browse Popular Pages:

No labels match these criteria.



This document provides general information only. ASX Limited (ABN 98 008 624 691) and its related bodies corporate (“ASX”) makes no representation or warranty with respect to the accuracy, reliability or completeness of the information. To the extent permitted by law, ASX and its employees, officers and contractors shall not be liable for any loss or damage arising in any way (including by way of negligence) from or in connection with any information provided or omitted or from anyone acting or refraining to act in reliance on this information.

© 2022 ASX Limited ABN 98 008 624 691