Use the Kaazing Gateway Java AMQP Client Library

In this procedure, you will learn how to use the Kaazing Gateway Java AMQP client library and the supported APIs. This topic shows you how to perform the following:

Set up your development environment
Review the common Java AMQP programming steps
Import the libraries
Import the Java AMQP classes
Declare the variables
Create the AmqpClient Object
Connect to an AMQP broker
Create channels
Declare an exchange
Declare a queue
Bind an exchange to a queue
Publish messages
Consume messages
Use transactions
Control message flow
Handle exceptions

Before You Begin

This procedure is part of Checklist: How to Build Java AMQP Clients.

Note: Learn about supported browsers, operating systems, and platform versions in the Release Notes.

For information about migrating your JavaScript AMQP 3.3-3.5 client to JavaScript 4.x, see Migrate Java AMQP Applications to the Kaazing AMQP Java 4.x Client SDK.

To Use the Kaazing Gateway Java AMQP Client Library

Set up your development environment.
1. To use a local Kaazing WebSocket Gateway, download and install the Gateway, as described in Setting Up the Gateway and Clients. If you simply want to test your client against a publicly available Kaazing WebSocket Gateway and AMQP broker, you can use the URL wss://demos.kaazing.com/amqp.
2. If you are using a local Kaazing WebSocket Gateway to develop clients, you must configure the Gateway to communicate with an AMQP broker.
  
  Note: If you have the Kaazing Gateway running on localhost and if you have an AMQP broker running on localhost at the default AMQP port 5672, you do not have to configure anything.
  
  The following is an example of the default configuration element for the AMQP service in the Kaazing WebSocket Gateway, as specified in the configuration file GATEWAY_HOME/conf/gateway-config.xml:
```
     
     <service>
     <accept>ws://localhost:8001/amqp</accept>
     <accept>wss://localhost:9001/amqp</accept>
     <connect>tcp://localhost:5672</connect>

     <type>amqp.proxy</type>
   
     <cross-site-constraint>
     <allow-origin>http://localhost:8001</allow-origin>
     </cross-site-constraint>
     <cross-site-constraint>
     <allow-origin>https://localhost:9001</allow-origin>
     </cross-site-constraint>
     </service>
    
```
  In this case, the service is configured to accept WebSocket AMQP requests from the browser at ws://localhost:8001/amqp and securely at wss://localhost:9001/amqp and proxy those requests to a locally installed AMQP broker (localhost) at port 5672.
  
  To configure the Gateway to accept WebSocket requests at another URL or to connect to a different AMQP broker, you can edit GATEWAY_HOME/conf/gateway-config.xml, update the values for the accept elements, change the connect property, and restart the Gateway. For example, the following configuration configures the Gateway to accept WebSocket AMQP requests at ws://www.example.com:80/amqp and proxy those requests to an AMQP broker (amqp.example.com) on port 5672.
```
     
     <service>
     <accept>ws://www.example.com:80/amqp</accept>
     <connect>tcp://amqp.example.com:5672</connect>
 
     <type>amqp.proxy</type>
     </service>
    
```
Review the common Java AMQP programming steps.

Now that you have set up your environment to develop Java applications using the Gateway's AMQP client library, you can start creating your application. You can either build a single application that both publishes and consumes messages, or create two different applications to handle each action. The tutorial located at https://github.com/kaazing/java.client.tutorials shows a single application that handles both actions.Refer to the AmqpClient Java API documentation for the complete list of all the AMQP command and callback functions.

Note: The Java AMQP programming examples listed in this topic use the Java code in the tutorial at https://github.com/kaazing/java.client.tutorials. Using the tutorial as an example when learning the AmqpClient Java API helps you to understand how the API classes are used in an application that captures and responds to user and message events.

The common Java AMQP programming steps are:
1. Import the client library
2. Import the Java AMQP classes
3. Declare the variables
4. Create the AmqpClient object
5. Connect to an AMQP broker
6. Create channels
7. Declare an exchange
8. Declare a queue
9. Bind an exchange to a queue
10. Publish messages
11. Consume messages
12. Use transactions
13. Control message flow
14. Handle exceptions
Import the libraries.

Import the Java client library from the options listed on kaazing.com/download. You can see an example of a build.gradle file using the libraries at https://github.com/kaazing/java.client.tutorials/blob/develop/j2se/java-amqp-demo/build.gradle.

Import the Java AMQP classes.

Add the following import statements in your application's .java file (the file JavaAmqpClientDemo.java is used in the Java AMQP demo):

package com.kaazing.amqp.client.demo;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.kaazing.net.ws.amqp.AmqpArguments;
import com.kaazing.net.ws.amqp.AmqpChannel;
import com.kaazing.net.ws.amqp.AmqpClient;
import com.kaazing.net.ws.amqp.AmqpClientFactory;
import com.kaazing.net.ws.amqp.AmqpProperties;
import com.kaazing.net.ws.amqp.ChannelAdapter;
import com.kaazing.net.ws.amqp.ChannelEvent;
import com.kaazing.net.ws.amqp.ConnectionEvent;
import com.kaazing.net.ws.amqp.ConnectionListener;

The highlighted classes (lines 15-21) are the relevant Java AMQP client library classes. The other classes are used to capture and respond to action events. In fact, the entire program is defined within the java.awt.event.ActionListener interface. See Lesson: Writing Event Listeners for more information.

Declare the variables.

In your application, first declare the variables you will use, as shown in the following example from the Java AMQP demo:

public class JavaAmqpClientDemo {
 private AmqpClient amqpClient;
 private AmqpChannel publishChannel = null;
 private AmqpChannel consumeChannel = null;
 private final String queueName = "queue" + new Random().nextInt();
 private final String exchangeName = "demo_exchange";
 private final String myConsumerTag = "clientkey";
 private final String routingKey = "broadcastkey";
 private final String virtualHost = "/";
 private String login;

Create the AmqpClient Object.

In the Java AMQP demo, the AmqpClient object is created when the program is loaded.

AmqpClientFactory amqpClientFactory = AmqpClientFactory.createAmqpClientFactory();
amqpClient = amqpClientFactory.createAmqpClient();

Connect to an AMQP broker.

Next, you must connect and log in to an AMQP broker using amqpClient.connect(). The client generally manages all of its communication on a single connection to an AMQP broker. You establish a connection to an AMQP broker by passing in the broker address, a username and password, the AMQP version you want to use, and, optionally, a virtual host name (the name of a collection of exchanges and queues hosted on independent server domains). These parameters are passed in when you call the amqpClient.connect() method as shown in the following example. The example from JavaAmqpClientDemo.java also demonstrates how to use CountDownLatch as a synchronization aid to allow one or more threads to wait until a set of operations is performed in other threads.

final CountDownLatch connectionLatch = new CountDownLatch(1);

amqpClient.addConnectionListener(new ConnectionListener() {

  public void onConnectionOpen(ConnectionEvent e) {
    System.out.println("CONNECTED...");
    connectionLatch.countDown();

  }

  public void onConnecting(ConnectionEvent e) {
    System.out.println("CONNECTING...");

  }

  public void onConnectionClose(ConnectionEvent e) {
    System.out.println("DISCONNECTING...");
    if (publishChannel != null) {
      publishChannel.closeChannel(0, "", 0, 0);
    }

    if (consumeChannel != null) {
      consumeChannel.closeChannel(0, "", 0, 0);
    }

  }

  public void onConnectionError(ConnectionEvent e) {
    System.err.println("CONNECTION ERROR! " + e.getMessage());
    System.exit(-1);
  }
});
amqpClient.connect(url.toString(), virtualHost, login, password);
connectionLatch.await(10, TimeUnit.SECONDS);

Note: The Gateway supports AMQP version 0-9-1.

Create channels.

Once a connection to an AMQP broker has been established, the client must create a channel to communicate to the broker. A channel is a bi-directional connection between an AMQP client and an AMQP broker. AMQP is multi-channeled, which means that channels are multiplexed over a single network socket connection. Channels are light-weight and consume little resources, and therefore used in AMQP's exception handling mechanism—channels are closed when an exception occurs.

You can create the channels as shown in the following example.
```
      publishChannel = amqpClient.openChannel();
      consumeChannel = amqpClient.openChannel();
```
Once you have created the channels, you can attach event handlers for when the connection opens.
```
publishChannel = amqpClient.openChannel();
publishChannel.addChannelListener(new ChannelAdapter() {
  @Override
  public void onClose(ChannelEvent e) {
    System.out.println("CLOSED: Publish Channel");
  }

  @Override
  public void onError(final ChannelEvent e) {
    System.err.println("ERROR: Publish Channel - " + e.getMessage());
    amqpClient.disconnect();
    System.exit(-1);
  }

  @Override
  public void onDeclareExchange(ChannelEvent e) {
    System.out.println("EXCHANGE DECLARED: " + exchangeName);
  }

  @Override
  public void onOpen(ChannelEvent e) {
    System.out.println("OPENED: Publish Channel");
    publishChannel.declareExchange(exchangeName, "fanout", false, false, false, null);
    pubChannelLatch.countDown();
  }
});
```
createChannel() uses ChannelAdapter() as part of an adapter or wrapper design pattern, and allows you to specify only the methods that your client requires, instead of having to specify every method. See Adapter pattern for more information on this design pattern.
Declare an exchange.

AMQP messages are published to exchanges. Messages contain a routing key that contains the information about the message's destination. The exchange accepts messages and their routing keys and delivers them to a message queue. You can think of an exchange as an electronic mailman that delivers the messages to a mailbox (the queue) based on the address on the message's envelope (the routing key). Exchanges do not store messages.

AMQP defines different exchange types. Some of these exchange types (Direct, Fanout, and Topic) must be supported by all AMQP brokers while others (Headers and System) are optional. AMQP brokers can also support custom exchange types. The following are the different types of exchanges:
- Direct: Messages are sent only to a queue that is bound with a binding key that matches the message's routing key.
- Fanout: Messages are sent to every queue that is bound to the exchange.
- Topic: Messages are sent to a queue based on categorical binding keys and wildcards.
- Headers: Messages are sent to a queue based on their header property values.
- System: Messages are sent to system services.
Exchanges can be durable, meaning that the exchange survives broker shutdown and must be deleted manually or non-durable (temporary), meaning that the exchange lasts only until the broker is shutdown. Finally, to check if an exchange exists on the AMQP broker (without actually creating it), you can create a passive exchange. The following example shows how you can create a direct exchange on the publish channel:
```
publishChannel.declareExchange("demo_exchange", &quot;fanout&quot;, false, false, false, null);
```
In this example, the exchange is demo_exchange, fanout is the exchange type, and false specifies whether the exchange is passive, durable, and noWait. null indicates that there are no AmqpArguments.
Declare a queue.

AMQP messages are consumed from queues. You can think of a queue as a mailbox: messages addressed to a particular address (the routing key) are placed in the mailbox for the consumer to pick up. If multiple consumers are bound to a single queue, only one of the consumers receives the message (the one that picked up the mail).

To check if a queue exists on the AMQP broker (without creating it), you can create a passive queue. Additionally, queues can be marked exclusive, meaning that they are tied to a specific connection. If a queue is marked exclusive, it is deleted when the connection on which it was created is closed.

Queues can be durable, meaning that the queue survives broker shutdown and must be deleted manually or non-durable (temporary) meaning that the queue lasts only until the broker is shut down. Queues can also be marked auto delete, meaning that the queue is deleted automatically when it is no longer in use. The following example shows how you can create a queue on the consume channel:
```
consumeChannel.declareQueue(queueName, false, false, false, false, false, null)
 .bindQueue(queueName, exchangeName, routingKey, false, null)
 .consumeBasic(queueName, myConsumerTag, false, false, false, false, null);
```
In this example, the queue value is obtained from the variable defined earlier:
```
private String queueName = "queue" + new Random().nextInt();
```
false specifies that the queue is not passive. durable, exclusive, autoDelete are not enabled, and noWait is not set. null indicates that there are no AmqpArguments.
Bind an exchange to a queue.

Once you have created an exchange and a queue in AMQP, you must bind—or map—one to the other so that messages published to a specific exchange are delivered to a particular queue. You bind a queue to an exchange with a routing key as shown in the following example.
```
consumeChannel.declareQueue(queueName, false, false, false, false, false, null)
 .bindQueue(queueName, exchangeName, routingKey, false, null)
 .consumeBasic(queueName, myConsumerTag, false, false, false, false, null);
```
After the exchange is bound to the queue successfully, a BindQueue event is raised, which writes the exchange and queue information to the log console:
```
public void onBindQueue(ChannelEvent e) {
 System.out.println("QUEUE BOUND: " + exchangeName + " - " + queueName);
}
```
Publish messages.

Messages are published to exchanges. The established binding rules (routing keys) then determine to which queue a message is delivered. Messages have content that consists of two parts:
1. Content Header: A set of properties that describes the message
2. Content Body: A blob of binary data
Additionally, messages can be marked mandatory to send a notification to the publisher in case a message cannot be delivered to a queue. You can also mark a message immediate so that it is returned to the sender if the message cannot be routed to a queue consumer immediately. The following example shows how the content body of a message is added to a buffer (AMQP uses a binary message format) and published to an exchange using the publish channel:
```
public void sendMessage(String message) {
	
 ByteBuffer buffer = ByteBuffer.allocate(512);
 buffer.put(message.getBytes(Charset.forName("UTF-8")));
 buffer.flip();

 Timestamp ts = new Timestamp(System.currentTimeMillis());
 AmqpProperties props = new AmqpProperties();
 props.setMessageId("1");
 props.setCorrelationId("4");
 props.setAppId("AMQPDemo");
 props.setUserId(this.login);
 props.setContentType("text/plain");
 props.setContentEncoding("UTF-8");
 props.setPriority(6);
 props.setDeliveryMode(1);
 props.setTimestamp(ts);

 AmqpArguments customHeaders = new AmqpArguments();
 customHeaders.addInteger("headerKey1", 100);
 customHeaders.addLongString("headerKey2", "Header value");

 props.setHeaders(customHeaders);

 publishChannel.publishBasic(buffer, props, exchangeName, routingKey, false, false);
 System.out.println("MESSAGE PUBLISHED: " + message);
}
```
A custom parameter is passed in for the message. The message text entered by the user is stored in a variable and converted to binary (buffer.putString(jTextField6.getText(), Charset.forName("UTF-8"));), and then sent to the exchange specified by the user (exchangeName). Also note that the last two arguments use the boolean value false for mandatory and immediate.

The AmqpProperties class defines pre-defined properties as per AMQP 0-9-1 spec and provides type-safe getters and setters for those pre-defined properties. The value of AMQP 0-9-1's standard "headers" property is of type AmqpArguments. The Kaazing Gateway AMQP implementation uses AmqpArguments to encode the table. Similarly, the Kaazing Gateway AMQP implementation decodes the table and constructs an instance of AmqpArguments.

The username set with the setUserId() method must match the user that is currently authenticated with the AMQP broker. If they do not match you will see the following error:
PRECONDITION_FAILED - user_id property set to '<name>' but authenticated user was '<name>'
Consume messages.

Once messages are published, they can be consumed from a queue. A variety of options can be applied to messages in a queue. For example, publishers can choose to require acknowledgement (ack) of messages so that messages can be redelivered in the case of a delivery failure. If the queue is set to exclusive, it is scoped to just the current connection and deleted when the connection on which it was established is closed. Additionally, you can use the no local setting to notify the broker not to send messages to the connection on which the messages were published. The following example shows how you can consume messages from a queue on the consume channel:
```
 consumeChannel.declareQueue(queueName, false, false, false, false, false, null)
 .bindQueue(queueName, jTextField5.getText(), routingKey, false, null)
 .consumeBasic(queueName, myConsumerTag, false, false, false, false, null);
```
After the consumeBasic method is successful, the AMQP broker can then start receiving messages in the client and these messages raise the Message event, which calls the corresponding event handler (onMessage()). The following example shows how the onMessage() function retrieves information from the event object (e):
```
public void onMessage(final ChannelEvent e) {
 byte[] bytes = new byte[e.getBody().remaining()];
 e.getBody().get(bytes);
 final Long dt = (Long) e.getArgument("deliveryTag");
 final String value = new String(bytes, Charset.forName("UTF-8"));

 System.out.println(">>> MESSAGE RECEIVED: " + value);
 AmqpProperties props = e.getAmqpProperties();
 if (props != null) {
  AmqpArguments headers = props.getHeaders();

  if (headers != null) {
   System.out.println("Headers: " + headers.toString());
  }
  System.out.println("Properties " + (String) props.toString());

  // Acknowledge the message as we passed in a 'false' for
  // noAck in AmqpChannel.consumeBasic() call. If the
  // message is not acknowledged, the broker will keep
  // holding the message. And, as more and more messages
  // are held by the broker, it will eventually result in
  // an OutOfMemoryError.
  AmqpChannel channel = e.getChannel();
  channel.ackBasic(dt.longValue(), true);
 }
}
```
Here you can see how the properties and headers are retrieved using AmqpProperties and AmqpArguments methods (getAmqpProperties() and getHeaders(), respectively).

Message Acknowledgement

The Boolean parameter noAck is optional with the default value of true. If noAck is true, the AMQP broker will not expect any acknowledgement from the client before discarding the message. If noAck is false, then the AMQP broker will expect an acknowledgement before discarding the message. If noAck is specified to be false, then you must explicitly acknowledge the received message using AmqpChannel ackBasic().

In the Java AMQP demo code in this procedure, message acknowledgement is being performed because false was passed in for noAck in consumeBasic(). If the Java client acknowledges a message and noAck is true (the default setting), then the AMQP message broker will close the channel.

Use transactions.

AMQP supports transactional messaging, through server local transactions. In a transaction, the server only publishes a set of messages as one unit when the client commits the transaction. Transactions only apply to message publishing and not to the consumption of the messages.

Note: Once you commit or rollback a transaction on a channel, a new transaction is started automatically. For this reason you must commit all future messages you want to publish on that channel or create a new, non-transactional channel to publish messages on.

The following transaction-related methods can be used to work select (start), commit, and rollback a transaction:

txnPublishChannel.selectTx();
txnPublishChannel.commitTx();
txnPublishChannel.rollbackTx();

After the transaction is selected successfully, committed, or rolled back, the corresponding events are raised and previously registered event handlers are called.

txnPublishChannel = amqpClient.openChannel();
txnPublishChannel.addChannelListener(new ChannelAdapter() {
  @Override
  public void onCommit(ChannelEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        jButton6.setEnabled(true);
        rollback.setEnabled(false);
        publish.setEnabled(false);
        jButton9.setEnabled(false);
        logMessage("TXN COMMITTED");
      }
    });
  }

  @Override
  public void onOpen(ChannelEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        // logMessage("OPENED: Publish Channel for Transaction");
        txnPublishChannel.declareExchange(jTextField7.getText(), 
        "fanout", false, false, false, null);
      }
    });
  }

  @Override
  public void onRollback(ChannelEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        jButton6.setEnabled(true);
        rollback.setEnabled(false);
        publish.setEnabled(false);
        jButton9.setEnabled(false);
        logMessage("TXN ROLLEDBACK");
      }
    });
  }
  @Override
  public void onSelect(ChannelEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        jButton6.setEnabled(false);
        logMessage("TXN SELECTED/STARTED");
      }
    });
  }
});
.
.
.

Control message flow.

You can use flow control in AMQP to temporarily—or permanently—halt the flow of messages on a channel from a queue to a consumer. If you turn the message flow off, no messages are sent to the consumer. The following example shows how you can turn the flow of messages on a channel off and back on:
```
consumeChannel.flowChannel(true);
consumeChannel.flowChannel(false);
```
After the flow on a channel is halted or resumed successfully, a flow event is raised, which calls the event handler registered previously.
```
public void onFlow(ChannelEvent e) {
  try {
    final boolean isActive = e.isFlowActive();
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        logMessage("FLOW: "+(isActive ? "ON" : "OFF"));
      }
    });
  }
  catch (Exception ex) {
    ex.printStackTrace();
  }
}
```

Handle exceptions.

Channels do not consume large resources, and therefore used in AMQP's exception handling mechanism—channels are closed when an exception occurs. In the Java AMQP demo, detailed information about the exception is captured using the onConnectionError() method for ConnectionListener and the onError() method for ChannelListener. Together, these methods can be used for all error handling.

  public void onConnectionError(final ConnectionEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        logMessage("ERROR:" + e.getMessage());
        status.setText("ERROR");
      }
    });
  }
  .
  .
  .
  public void onError(final ChannelEvent e) {
    EventQueue.invokeLater(new Runnable() {
      @Override
      public void run() {
        logMessage("ERROR: Publish Channel - " + e.getMessage());
      }
    });
  }

Migrate Java AMQP Applications to the Kaazing AMQP Java 4.x Client SDK

If you wish to migrate your Kaazing Gateway 3.3-3.5 Java AMQP clients to the Kaazing AMQP Java 4.x Client SDK and use its new libraries, do the following:

Kaazing Gateway 3.3-3.5 Package	Kaazing AMQP Java 4.x Client SDK
com.kaazing.gateway.amqp.client	com.kaazing.net.ws.amqp
com.kaazing.gateway.client	com.kaazing.net.ws
com.kaazing.gateway.client.security	com.kaazing.net.auth

Use the new WebSocket library in your client, as described above.
Use the new Java AMQP Client library, as described above.
Update your Java AMQP client to use the new package names: com.kaazing.gateway.amqp.client is now com.kaazing.net.ws.amqp. For example, import com.kaazing.gateway.amqp.client.AmqpClient is now com.kaazing.net.ws.amqp.AmqpClient. Also, the security package names have changed. For example, com.kaazing.gateway.client.security.LoginHandler is now com.kaazing.net.auth.LoginHandler.

Update your Java AMQP client to use the new factory class:

Kaazing Gateway 3.3-3.5:

import com.kaazing.gateway.amqp.client.AmqpClient;
...
private AmqpClient amqpClient;
...
public void actionPerformed(ActionEvent arg0) {
  if (arg0.getSource() == connect) {
      try {
          amqpClient = new AmqpClient();
          ...

Kaazing AMQP Java 4.x Client SDK:

import com.kaazing.net.ws.WebSocketFactory;
import com.kaazing.net.ws.amqp.AmqpClient;
import com.kaazing.net.ws.amqp.AmqpClientFactory;
...
private AmqpClientFactory amqpClientFactory;
private AmqpClient mqpClient;
...
amqpClientFactory = AmqpClientFactory.createAmqpClientFactory();
...
    public void actionPerformed(ActionEvent arg0) {
        if (arg0.getSource() == connect) {
            try {
                amqpClient = amqpClientFactory.createAmqpClient();
                WebSocketFactory wsFactory = amqpClient.getAmqpClientFactory().getWebSocketFactory();
                ...

Modify challenge handlers. In Kaazing Gateway 4.x, the ChallengeHandlers class from 3.3-3.5 was replaced with by the ChallengeHandler modifier of the WebSocketFactory class. The ChallengeHandler modifier is used during authentication for connections and subsequent revalidation that occurs at regular intervals. Also, note the new package names.