This guide contains information about the AMPS JavaScript client.

It focuses on information that is specific to the AMPS JavaScript client. The guide does not contain detailed information on AMPS itself.

To get started with AMPS, see the Introduction to AMPS guide. That guide contains an overview of AMPS and instructions for setting up a development environment.

Manual Installation

The AMPS JavaScript client is available as a download from the 60East Technologies web site. Download the client from the site, then extract it.

The client source files are in the directory where you unpacked the files. By default, this is amps-javascript-client-<version>, where <version> is the current version of the JavaScript client, such as amps-javascript-client-5.2.1.1).

Once unpacked, amps.js can be included in the project in order to use the client. Optionally, depending on your development environment, the es6-promise.js might also be required.

Installation via Node Package Manager

If your project is using Node Package Manager to manage and install dependencies, the client can be downloaded and installed via NPM:

npm install --save amps

The client library will be automatically installed and included in your project.

Introduction

Welcome to developing applications with AMPS, the Advanced Message Processing System from 60East Technologies!

These guides will help you learn how to develop applications using AMPS.

Prerequisites

Before reading this guide, it is important to have a good understanding of the following topics:

• Developing in JavaScript - To be successful using this guide, you will need to possess a working knowledge of the JavaScript language. Visit https://developer.mozilla.org/en-US/docs/Learn/JavaScript for resources on learning JavaScript.

• AMPS concepts - Before reading this guide, you will need to understand the basic concepts of AMPS, such as topics, subscriptions, messages and SOW.

  Before working through this guide, we recommend reading the Introduction to AMPS guide.

  Detailed explanations of the AMPS server behavior are in the AMPS Server Documentation.

• An installed browser or Node.js runtime - The AMPS JavaScript Client currently supports most JavaScript environments.

You will also need a system on which you can compile and run code, and a server where you can host the AMPS server.

JavaScript Support

The AMPS JavaScript client supports both Browser and Node.js as well as their derivatives, such as Electron, and OpenFin. The supported environments are the following:

• Node.js 0.12.1 or higher;

• Microsoft Internet Explorer 11;

• Microsoft Edge 12 or higher;

• Google Chrome 16 or higher;

• Mozilla Firefox 11 or higher;

• Apple Safari 6.1 (iOS 6.0) or higher;

• Apple iOS Safari 7 or higher;

• Google Android Browser 4.4 (Android KitKat) or higher;

• Opera 12.1 or higher.

The JavaScript client supports all API features.

Setting up a Development Instance

You will need an installed and running AMPS server to use the product as well. You can write and compile programs that use AMPS without a running server, but you will get the most out of this guide by running the programs against a working server.

Instructions for starting an instance of AMPS are available in the Introduction to AMPS guide.

AMPS uses the name of the client as a session identifier and as part of the identifier for messages originating from that client.

For this reason, when a transaction log is enabled in the AMPS instance (that is, when the instance is recording a sequence of publishes and attempting to eliminate duplicate publishes), an AMPS instance will only allow one application with a given client name to connect to the instance.

When a transaction log is present, AMPS requires the client name for a publisher to be:

• Unique within a set of replicated AMPS instances

• Consistent from invocation to invocation if the publisher will be publishing the same logical stream of messages

If publishers do not meet this contract (for example, if the publisher changes its name and publishes the same messages, or if a different publisher uses the same session name), message loss or duplication can happen.

60East recommends always using consistent, unique client names. For example, the client name could be formed by combining the application name, an identifier for the host system, and the ID of the user running the application. A strategy like this provides a name that will be different for different users or on different systems, but consistent for instances of the application that should be treated as equivalent to the AMPS system.

Likewise, if a publisher is sending a completely independent stream of messages (for example, a microservice that sends a different, unrelated sequence of messages each time it connects to AMPS), there is no need for a publisher to retain the same name each time it starts. However, if a publisher is resuming a stream of messages (as in the case when using a file-backed publish store), that publisher must use the same client name, since the publisher is resuming the session.

The AMPS clients use connection strings to determine the server, port, transport, and protocol to use to connect to AMPS. When the connection point in AMPS accepts multiple message types, the connection string also specifies the precise message type to use for this connection.

Connection strings have a number of elements:

As shown in the figure above, connection strings have the following elements:

• Transport - Defines the network used to send and receive messages from AMPS. In this case, the transport is wss.

  ws (WebSocket) and wss (WebSocket Secure) are the only supported transports for the JavaScript client.

• Host Address - Defines the destination on the network where the AMPS instance receives messages. The format of the address is dependent on the transport. For ws and wss, the address consists of a host name and port number. In this case, the host address is localhost:9007.

• Protocol - Sets the format in which AMPS receives commands from the client. Most code uses the default amps protocol, which sends header information in JSON format. AMPS supports the ability to develop custom protocols as extension modules, and AMPS also supports legacy protocols for backward compatibility.

• Message Type - Specifies the message type that this connection uses. This component of the connection string is required if the protocol accepts multiple message types and the transport is configured to accept multiple message types. If the protocol does not accept multiple message types, this component of the connection string is optional, and defaults to the message type specified in the transport.

  Legacy protocols such as fix, nvfix and xml only accept a single message type, and therefore do not require or accept a message type in the connection string.

The JavaScript client does not support legacy protocols, the amps protocol is required.

As an example, a connection string such as:

ws://localhost:9007/amps/json

would work for programs connecting from the local host to a Transport configured as follows:

<AMPSConfig>
...

<Protocols>
    ...

    <!-- This protocol is using the websocket module,which is
    required for JavaScript client. -->
    <Protocol>
        <Name>websocket-any</Name>
        <Module>websocket</Module>
    </Protocol>

</Protocols>

...

<Transports>
    ...

    <!-- This transport accepts any known message type for
    the instance: the client must specify the message type. -->
    <Transport>
        <Name>websocket-any</Name>
        <Type>tcp</Type>
        <InetAddr>9007</InetAddr>
        <Protocol>websocket-any</Protocol>
    </Transport>

</Transports>

...

</AMPSConfig>

See the AMPS Configuration Guide for more information on configuring transports and protocols.

So far, we have seen that subscribing to a topic involves working with objects of the Message type. A Message represents a single message from an AMPS server, while a Command is sent to a server. Commands are sent and messages are received for every client/server operation in AMPS.

Header Properties

There are two parts of each message in AMPS: a set of headers that provide metadata for the message, and the data that the message contains. Every AMPS message has one or more header fields defined. The precise headers present depend on the type and context of the message. There are many possible fields in any given message, but only a few are used for any given message. For each header field, the Message object contains a distinct method that allows for retrieval of that field. For example, the Message.header.commandId() corresponds to the CommandId header field, the Message.header.batchSize() corresponds to the BatchSize header field, and so on. For more information on these header fields, consult the AMPS User Guide and AMPS Command Reference.

Message class represents messages received from an AMPS server. When creating a message to be sent, the Command class is used. To work with header fields, a Command contains a set of <propertyName>() methods, which work as both setters and getters, allowing to chain command properties when creating a new command.

Message Data

Received message data is contained in the Message.data property. The data property will contain the parsed data of the message.

The AMPS JavaScript client contains a collection of helper classes for working with message types that are specific to AMPS (for example, JSON, FIX, NVFIX, and AMPS composite message types). You can replace default parsers to implement required specific behavior, as well as add new helpers.

AMPS allows you to update parameters, such as the content filter, on a subscription. When you replace a filter on the subscription, AMPS immediately begins sending only messages that match the updated filter. Notice that if the subscription was entered with a command that includes a SOW query, using the replace option can re-issue the SOW query (as described in the AMPS User Guide).

To update the filter on a subscription, you create a subscribe command. You set the SubscriptionId provided on the Command to the identifier of the existing subscription and include the replace option on the Command.

When you send the Command, AMPS atomically replaces the filter and sends messages that match the updated filter from that point forward.

await client.connect('wss://localhost:9000/amps/json');

// Original subscription
const subId = await client.execute(
  new Command('subscribe').topic('orders').filter('/id > 100'),
  onMessagePrinter
);

// replace the subscription with a new filter value
await client.execute(
  new Command('subscribe')
    .topic('orders')
    .filter('/id < 5')
    .subId(subId)
    .options('replace'),
  onMessagePrinter
);

Regular Expression (Regex) subscriptions allow a regular expression to be supplied in the place of a topic name. When you supply a regular expression, it is as if a subscription is made to every topic that matches your expression, including topics that do not yet exist at the time of creating the subscription.

To use a regular expression, simply supply the regular expression in place of the topic name in the subscribe() call. For example:

await client.subscribe(message => { ... }, 'orders.*');

In this example, messages on topics `orders-north-america`,
`orders-europe`, and `new-orders` would match the regular expression,
and those messages would all be sent to the message handler function.
As in the example, you can use the `message.header.topic()` method to
determine the actual topic of the message sent to the function.

When using the default authenticator, the AMPS clients support the standard format for including a username and password in a URI, as shown below:

ws://user:password@host:port/protocol/message_type

When provided in this form, the default authenticator provides the username and password specified in the URI. If you have implemented another authenticator, that authenticator controls how passwords are provided to the AMPS server.

With asynchronous message processing, when a subscription is successfully made, messages will begin flowing to the message handler function and the Client.subscribe() call returns a Promise object that resolves with a unique string that serves as the identifier for this subscription. A Client can have any number of active subscriptions, and this string is used to refer to the particular subscription we have made here. For example, to unsubscribe, we simply pass in this identifier:

const client = new Client('exampleClient');
await client.connect('wss://localhost:9000/amps/json');

const subId = await client.subscribe(onMessagePrinter, 'messages');

// when the program is done with the subscription, unsubscribe
await client.unsubscribe(subId);
console.log('Unsubscribed');

In this example, as in the previous section, we use the Client.subscribe() method to create a subscription to the messages topic. When our application is done listening to this topic, it unsubscribes by passing in the subId passed from the successfully resolved Promise of subscribe(). After the subscription is removed, no more messages will flow into our onMessagePrinter function.

AMPS also accepts the keyword all to unsubscribe all subscriptions for the client.

The first time a command causes an instance of the Client to connect to AMPS, the client creates a WebSocket connection that runs asynchronously. This asynchronous connection is responsible for processing incoming messages from AMPS, which includes both messages that contain data and acknowledgments from the server.

When you execute a command on the AMPS client, the execute() method creates and returns a Promise object that typically waits for an acknowledgment from the server and then resolves with the id of the command. (The exception to this is Client.publish(). For performance, the publish command does not wait for an acknowledgment from the server before returning.)

Message handlers provided for message processing must be aware of the following considerations:

• For maximum performance, do as little work in the message handler as possible. For example, if you use the contents of the message to perform an extensive calculation, a message handler that passes data into a WebWorker instance will typically perform better than a message handler that does this calculation in place.

• While your message handler is running, the connection that calls your message handler is no longer receiving messages. This makes it easier to write a message handler because you know that no other messages are arriving from the same subscription. However, this also means that you cannot use the same client that called the message handler to send commands to AMPS. Acknowledgments from AMPS cannot be processed and your application will block while waiting for the acknowledgment. Instead, enqueue the command in a work queue to be processed by a separate worker or use a different client object to submit the commands.

In every distributed system, the robustness of your application depends on its ability to recover gracefully from unexpected events. The AMPS client provides the building blocks necessary to ensure your application can recover from the kinds of errors and special events that may occur when using AMPS.

Every distributed system will experience occasional disconnections between one or more nodes. The reliability of the overall system depends on an application's ability to efficiently detect and recover from these disconnections. Using the AMPS JavaScript client's disconnect handling, you can build powerful applications that are resilient in the face of connection failures and spurious disconnects. For additional reliability, you can also use the high availability functionality (discussed in the following sections), which provides both disconnect handling and features to help ensure that messages are reliably delivered.

One of the most powerful features of AMPS is content filtering. With content filtering, filters based on message content are applied at the server so that your application and the network are not utilized by messages that are not relevant to your application. For example, if your application is only displaying messages from a particular user, you can send a content filter to the server so that only messages from that particular user are sent to the client.

To apply a content filter to a subscription, simply pass it into the Client.subscribe() call or use the filter() method to add a filter to the Command:

await client.subscribe(
  msg => console.log('Mom says: ', msg.data),
  'letters',         // Topic
  '/sender="mom"',   // Filter
  {timeout: 5000}    // Timeout (optional)
);

In this example, we have passed in a content filter /sender = 'mom'. This will result in the server only sending us messages, from the messages topic, that have the sender field equal to mom in the message.

For example, the AMPS server will send the following message, where /sender is mom:

{
    "sender" : "mom",
    "text" : "Happy Birthday!",
    "reminder" : "Call me Thursday!"
}

The AMPS server will not send a message with a different /sender value:

{
    "sender" : "henry dave",
    "text" : "Things do not change; we change."
}

In this chapter, we will learn more about the structure and features of the AMPS JavaScript client and build our first JavaScript program using AMPS.

About the Client Library

The AMPS client library is packaged as a single file, amps.js in development environments, or amps.min.js for production. Every JavaScript application you build will need to reference this file, and the file must be deployed along with your application in order for your application to function properly. Alternatively, for applications that are distributed through NPM, the client library can be listed as a dependency, and thus installed automatically.

Including the Client in a Project

There are several ways of including the JavaScript client in order to use it. Depending on your project's structure and environment, you can include the client in one of the following ways.

Node.js / CommonJS Module

// NPM installation: Import everything at once
const amps = require('amps');

// or

// NPM installation: Include only selected components of AMPS
const { Client, Command } = import('amps');

// Same as above with manual installation
const amps = require('./lib/amps');
const { Client, Command } = import('./lib/amps');

ES6 / TypeScript Module

// NPM installation
import { Client, Command } from 'amps';

// Manual installation
import { Client, Command } from './lib/amps';

In case of TypeScript and manual installation, you need to reference the typing file (provided within the client package) at the top of the file:

/// <reference path="./lib/amps.d.ts" />
import { Client, Command } from './lib/amps';

The AMPS JavaScript NPM package already includes typings, so that the library can be seamlessly used in native TypeScript projects.

AMD / RequireJS Module

// Included library will be available as the amps argument
define(['./lib/amps'], amps => {
    // ...
});

Global Import in a Browser Environment

<!-- Optional import, to support obsolete browsers like IE11 -->
<script src="/lib/es6-promise.min.js"></script>
<script src="/lib/amps.js"></script>

Once the library in included in the HTML file, it becomes available as a global object, either as amps or window.amps.

Connecting to AMPS

Let's begin by writing a simple program that connects to an AMPS server and sends a single message to a topic. This code will be covered in detail just following the example.

const uri = 'wss://127.0.0.1:9007/amps/json';

async function main() {
    const client = new Client('examplePublisher');

try {
    await client.connect(uri);
    client.publish('messages', {hi: 'Hello, World!'});
    client.disconnect();
  }

catch (err) {
    console.error(err);
  }
}

main();

In the example above, we show the entire program; but future examples will isolate one or more specific portions of the code.

Examining the Code

Let us now revisit the code we listed above:

/**
 * The URI to use to connect to AMPS. The URI consists of
 * the transport, the address, and the protocol to use for
 * the AMPS connection. In this case, the transport is `wss`
 * (WebSocket Secure), the address is `127.0.0.1:9007`, and
 * the protocol is `amps`. This connection will be used for
 * JSON messages. Check with the person who manages the AMPS
 * instance to get the connection string to use for your programs.
 */
const uri = 'wss://127.0.0.1:9007/amps/json';


async function main() {
  /**
   * This line creates a new Client object. Client encapsulates
   * a single connection to an AMPS server. Methods on Client allow
   * for connecting, disconnecting, publishing, and subscribing to
   * an AMPS server. The argument to the Client constructor,
   * `examplePublisher`, is a name chosen by the client to identify
   * itself to the server. Errors relating to this connection will be
   * logged with reference to this name, and AMPS uses this name to
   * help detect duplicate messages. AMPS enforces uniqueness for
   * client names when a transaction log is configured, and it is
   * good practice to always use unique client names.
   */
  const client = new Client('examplePublisher');

  /**
   * Now, we attempt to connect to the AMPS server. Most API methods
   * use the Promise pattern to encapsulate asynchronous operations,
   * such as connecting or subscribing. We utilize the async/await
   * syntax to work with the asynchronous code in a somewhat
   * synchronous manner.
   * For example, once the connection is established, the publish
   * function will be called; if the call to connect() fails because
   * the provided address is not reachable, an error will be thrown
   * and caught in the `catch` clause with an Error object that
   * contains information about the occurred error.
   */
  try {
    /**
    * Here, we provide a URI string to the client and establish
    * connection to AMPS.
    */
    await client.connect(uri);

    /**
    * Here, we publish a single message to AMPS on the messages
    * topic, containing the data `Hello, world!`. This data is
    * placed into a JSON message and sent to the server. Upon
    * successful completion of this function, the AMPS client has
    * sent the message to be sent to the server, and subscribers
    * to the messages topic will receive this JSON message:
    * '{"hi": "Hello, world!"}'.
    */
    client.publish('messages', {hi: 'Hello, World!'});

    /**
    * We close the connection to AMPS. It's good practice to
    * close connections when you are done using them as the
    * client object will stay connected otherwise.
    */
    client.disconnect();
  }
  catch (err) {
    // Failed connection error
    console.error(err);
  }
}


// running the above function
main();

Syntax Convention

In the above example and throughout this guide we use the modern JavaScript syntax which is widely supported and typically is used in projects with a build/deployment system, such as webpack, angular-cli, and create-react-app. However, the JavaScript client library fully supports the obsolete syntax used in older browsers and projects without a build system as well. Below is the same program, written in the classic ES5 syntax.

var uri = 'wss://127.0.0.1:9007/amps/json';
var client = new amps.Client('examplePublisher');

client.connect(uri)
    .then(function() {
        client.publish('messages', {hi: 'Hello, world!'});

        client.disconnect();
    })
    .catch(function(error) {
        console.error(error);
    });

The AMPS client includes a heartbeat feature to help applications detect disconnection from the server within a predictable amount of time. Without using a heartbeat, an application must rely on the environment to notify it when a disconnect occurs. For applications that are simply receiving messages, it can be impossible to tell whether a socket is disconnected or whether there are simply no incoming messages for the client.

When you set a heartbeat, the AMPS client sends a heartbeat message to the AMPS server at a regular interval, and waits a specified amount of time for the response. If the operating system reports an error on send, or if the server does not respond within the specified amount of time, the AMPS client considers the server to be disconnected.

The AMPS client processes heartbeat messages asynchronously. If your application publishes messages in a loop (a synchronous operation), or the message handler is receiving significant amount of messages at the moment, the client may fail to respond to heartbeat messages in a timely manner and may be disconnected by the server.

// Create a new client and set the heartbeat interval to 5 sec
const client = new Client('heartbeat-demo').heartbeat(5);

The Client class, included with the AMPS JavaScript client, contains a disconnect handler and other features for building highly-available applications. The Client includes features for managing a list of failover servers, resuming subscriptions, republishing in-flight messages, and other functionality that is commonly needed for high availability. This section covers the use of a custom disconnect handler in the event that the behavior of the Client does not suit the needs of your application. 60East recommends using the provided disconnect handler unless there is specific behavior that does not meet your needs (for example, if your application does not want to reconnect to AMPS in the event of a disconnection).

In some cases, an application does not want the AMPS Client to reconnect, but instead wants to take a different action if disconnection occurs. For example, a stateless publisher that sends ephemeral data (such as telemetry or prices) may want to exit with an error if the connection is lost rather than risk falling behind and providing outdated messages. Often, in this case, a monitoring process will start another publisher if a publisher fails, and it is better for a message to be lost than to arrive late.

To cover cases where the application has unusual needs, the AMPS client library allows an application to provide custom disconnect handling.

Your application gets to specify exactly what happens when a disconnect occurs by supplying a function to client.setDisconnectHandler(), which is invoked whenever a disconnect occurs. This may be helpful for situations where a particular connection needs to do something completely different than reconnecting or failing over to another AMPS server.

The example below shows the basics:

/**
* Call this function to establish a connection to AMPS.
*/
const connectToAMPS = async () => {
  try {
    await client.connect('ws://localhost:9000/amps/json');
    // Successfully connected
  }
  catch (err) {
    // Can't establish connection
  }
};

// create a client object
const client = new Client('disconnect-handler-demo');

/*
* disconnectHandler() method is called to supply a function for use
* when AMPS detects a disconnect. At any time, this function may be
* called by AMPS to indicate that the client has disconnected from
* the server, and to allow your application to choose what to do
* about it.
*/
client.disconnectHandler(
    /**
    * Our disconnect handler’s implementation begins here.
    *
    * Any custom disconnect handler would be application-specific
    * so, for demonstration purposes, we simply log the error.
    *
    * Notice that this disconnect handler replaces all other
    * disconnect handling behavior. When the client is disconnected,
    * it will simply log an error to the console. The client
    * will not reconnect or take any other action.
    */
    (client, error) => {
         console.log(error);
    }
);

// We begin by connecting and subscribing
connectToAMPS();

Messages published to a topic on an AMPS server are available to other clients via a subscription. Before messages can be received, a client must subscribe to one or more topics on the AMPS server so that the server will begin sending messages to the client. The server will continue sending messages to the client until the client unsubscribes, or the client disconnects. With content filtering, the AMPS server will limit the messages sent to only those messages that match a client-supplied filter. In this chapter, you will learn how to subscribe, unsubscribe, and supply filters for messages using the AMPS JavaScript client.

Subscribing to a Topic

The AMPS client makes it simple to subscribe to a topic. You call Client.subscribe() with the message handler, the topic to subscribe to and the parameters for the subscription. The client submits the subscription to AMPS and returns a Promise object that resolves with the subscription identifier. Received messages are asynchronously delivered to the message handler function. Below is a short example:

/**
* Let's create a message handler function. It will be invoked
* when AMPS delivers messages to the subscription.
* Within this handler, we process messages. In this case,
* we simply print the contents of the message.
*/
const onMessage = message => console.log(message.data);

// let's create a Client that is connected to an AMPS server.
const client = new Client('test');
await client.connect('wss://127.0.0.1:9007/amps/json');

/**
* Here we subscribe to the topic "messages". We do not provide
* a filter, so AMPS does not content-filter the topic.
* The command resolves with the id of the subscription.
*/
const subId = await client.subscribe(onMessage, 'messages');
console.log('subId:', subId);

AMPS creates an asynchronous subscription that receives messages and calls the message handler only if there's a new message. This means that the client application as a whole can continue to receive messages while you are doing processing work, by stacking them in the execution queue.

The simple method described above is provided for convenience. The AMPS JavaScript client provides convenience methods for the most common form of AMPS commands. The client also provides an interface that allows you to have precise control over the command. Using that interface, the example above becomes:

/**
* Let's create a message handler function. It will be invoked
* when AMPS delivers messages to the subscription.
* Within this handler, we process messages. In this case,
* we simply print the contents of the message.
*/
const onMessage = message => console.log(message.data);

// let's create a Client that is connected to an AMPS server.
const client = new Client('test');
await client.connect('wss://127.0.0.1:9007/amps/json');

/**
* Here we create a Command object for the subscribe
* command, specifying the topic "messages". We do not
* provide a filter, so AMPS does not content-filter
* the topic.
*/
const cmd = new Command('subscribe').topic('messages');

/**
* We execute the Command using the execute() method.
* The command execution resolves with the id of the
* subscription.
*/
const subId = await client.execute(cmd, onMessage);
console.log('subId: ', subId);

The Command interface allows you to precisely customize the commands you send to AMPS. For flexibility and ease of maintenance, 60East recommends using the Command interface (rather than a named method) for any command that will receive messages from AMPS. For publishing messages, there can be a slight performance advantage to using the named commands where possible.

When using the asynchronous interface, errors can occur that are not thrown to the user. For example, when a message with invalid JSON data was received, the error occurs in the process of parsing its data inside of the AMPS JavaScript Client. Consider the following example:

const onMessage = message => console.log(message.data);
await client.subscribe(onMessage, 'pokes', '/Pokee LIKE ' + userId);

In this example, we set up a subscription to wait for messages on the pokes topic, whose Pokee tag begins with our user ID. When messages arrive, we print a message out to the console.

Inside of the AMPS client, the client received an message that contains invalid JSON data that cannot be parsed. When the error occurs, there is no handler for it to be reported to, and by default it is ignored.

In applications where it is important to deal with every issue that occurs in using AMPS, you can set an error handler via Client.errorHandler() that receives these otherwise-unhandled errors and exceptions. Making the modifications shown in the example below, to our previous example, will allow those errors to be caught and handled. In this case we are simply printing those caught errors out to the console.

const onMessage = message => console.log(message.data);

// This time, let's add the error handler
client.errorHandler(err => console.error('Error occurred:', err));

await client.subscribe(onMessage, 'pokes', '/Pokee LIKE ' + userId);

In this example we have added a call to client.errorHandler(), registering a simple function that writes the text of the error out to the console. If errors are thrown in the message handler, those errors are written to the console.

The Client class contains a disconnect handler and other features for building highly-available applications. The Client includes features for managing a list of failover servers, resuming subscriptions, republishing in-flight messages, and other functionality that is commonly needed for high availability. 60East recommends using the Client for automatic reconnection wherever possible, as the disconnect handler has been carefully crafted to handle a wide variety of edge cases and potential failures.

If an application needs to reconnect or fail over, use a Client with a ServerChooser set, and the AMPS client library will automatically handle failover and reconnection. You control which servers the client fails over to by implementing the ServerChooser and you can control the timing of the failover by using one of the provided ReconnectDelayStrategy classes or implementing your own.

If you need to add custom behavior to the failover (such as logging, resetting an internal cache, refreshing credentials and so on), the ConnectionStateListener class allows your application to be notified and take action when disconnection is detected and at each stage of the reconnection process.

To extend the behavior of the AMPS client during reconnection, implement a ConnectionStateListener.

The AMPS clients include a batch size parameter that specifies how many messages the AMPS server will return to the client in a single batch when returning the results of a SOW query. The 60East clients set a batch size of 10 by default. This batch size works well for common message sizes and network configurations.

Adjusting the batch size may produce better network utilization and produce better performance overall for the application. The larger the batch size, the more messages AMPS will send to the network layer at a time. This can result in fewer packets being sent, and therefore less overhead in the network layer. The effect on performance is generally most noticeable for small messages, where setting a larger batch size will allow several messages to fit into a single packet. For larger messages, a batch size may still improve performance, but the improvement is less noticeable.

In general, 60East recommends setting a batch size that is large enough to produce few partially-filled packets. Bear in mind that AMPS holds the messages in memory while batching them, and the client must also hold the messages in memory while receiving the messages. Using batch sizes that require large amounts of memory for these operations can reduce overall application performance, even if network utilization is good.

For smaller message sizes, 60East recommends using the default batch size, and experimenting with tuning the batch size if performance improvements are necessary. For relatively large messages (especially messages with sizes over 1MB), 60East recommends explicitly setting a batch size of 1 as an initial value, and increasing the batch size only if performance testing with a larger batch size shows improved network utilization or faster overall performance.

AMPS State of the World (SOW) allows you to automatically keep and query the latest information about a topic on the AMPS server, without building a separate database. Using SOW lets you build impressively high-performance applications that provide rich experiences to users. The AMPS JavaScript client lets you query SOW topics and subscribe to changes with ease.

Performing SOW Queries

To begin, we will look at a simple example of issuing a SOW query.

const onMessage = message => {
  switch (message.header.command()) {
    case 'group_begin':
      console.log('--- Begin SOW Results ---');
      break;

    case 'sow':
      console.log(message.data);
      break;

    case 'group_end':
      console.log('--- End SOW Results ---');
      break;
  }
};

const queryId = await client.sow(
  onMessage,       // Message handler
  'orders',        // SOW Topic
  '/symbol="ROL"'  // Filter
);

In the example above, we invoke Client.sow() to initiate a SOW query on the orders topic, for all entries that have a symbol of 'ROL'. As usual, the Client.sow() method returns a Promise object that resolves with the query ID.

As the query executes, the message handler function is invoked for each matching entry in the topic. Messages containing the data of matching entries have a Command of value sow, so as those arrive, we write them to the console.

AMPS allows applications to manage the contents of the SOW by explicitly deleting messages that are no longer relevant. For example, if a particular delivery van is retired from service, the application can remove the record for the van by deleting the record for the van.

The client provides the following methods for deleting records from the SOW:

• sowDelete() - Accepts a topic and filter, and deletes all messages that match the filter from the topic specified.

• sowDeleteByKeys() - Accepts a set of SOW keys as a comma-delimited string and deletes messages for those keys, regardless of the contents of the messages. SOW keys are provided in the header of a SOW message, and are the internal identifier AMPS uses for that SOW message.

• sowDeleteByData() - Accepts a topic and message, and deletes the SOW record that would be updated by that message.

Most applications use sowDelete(), since this is the most useful and flexible method for removing items from the SOW. In some cases, particularly when working with extremely large SOW databases, sowDeleteByKeys() can provide better performance.

In either case, AMPS sends an OOF message to all subscribers who have received updates for the messages removed, as described in the previous section.

sowDelete() returns a Promise that resolves with a Message object. This Message is an acknowledgment that contains information on the delete command. For example, the following snippet simply prints informational text with the number of messages deleted:

const ack = await client.sowDelete('sow-topic', '/id IN (42, 37)');

console.log(
  'Got an',
  ack.header.command(), 'message containing',
  ack.header.ackType(), '-- deleted',
  ack.header.matches(), 'SOW entries'
);

// Got an ack message containing stats -- deleted 2 SOW entries

Acknowledging messages from a queue uses a form of the sow_delete command that is only supported for queues. Acknowledgment is discussed in the Using Queues chapter in this guide.

AMPS message queues provide a high-performance way of distributing messages across a set of workers. The AMPS User Guide describes AMPS queues in detail, including the features of AMPS referred to in this chapter. This chapter does not describe AMPS queues in detail, but instead explains how to use the AMPS JavaScript client with message queues.

To publish messages to a message queue, publishers simply publish to any topic that is collected by the queue. There is no difference between publishing to a queue and publishing to any other topic, and a publisher does not need to be aware that the topic will be collected into a queue.

Subscribers must be aware that they are subscribing to a queue, and acknowledge messages from the queue when the message is processed.

The AMPS JavaScript client automatically batches acknowledgments when either of the convenience methods is used. Batching acknowledgments reduces the number of round-trips to AMPS, which reduces network traffic and improves overall performance. AMPS sends the batch of acknowledgments when the number of acknowledgments exceeds a specified size, or when the amount of time since the last batch was sent exceeds a specified timeout.

You can set the number of messages to batch and the maximum amount of time between batches, as shown below:

client.ackBatchSize(10);  // Send batch after 10 messages
client.ackTimeout(1000);  // ... or 1 second

The AMPS JavaScript client is aware of the subscription backlog for a subscription. When AMPS returns the acknowledgment for a subscription that contains queues, AMPS includes information on the subscription backlog for the subscription. If the requested batch size is larger than the subscription backlog, the AMPS JavaScript client adjusts the requested batch size to match the subscription backlog.

Acknowledging Messages

For each message delivered on a subscription, AMPS counts the message against the subscription backlog until the message is explicitly acknowledged. In addition, when a queue specifies at-least-once delivery, AMPS retains the message in the queue until the message expires or until the message has been explicitly acknowledged and removed from the queue. From the point of view of the AMPS server, acknowledgment is implemented as a sow_delete from the queue with the bookmarks of the messages to remove. The AMPS JavaScript client provides several ways to make it easier for applications to create and send the appropriate sow_delete.

Automatic Acknowledgment

The AMPS client allows you to specify that messages should be automatically acknowledged. When this mode is on, AMPS acknowledges the message automatically if the message handler returns without throwing an exception.

AMPS batches acknowledgments created with this method, as described in the following section.

To enable automatic acknowledgment, use the Client.autoAck() method.

client.autoAck(true);  // enable AutoAck

Message Convenience Method

The AMPS JavaScript client provides a convenience method, Client.ack(), on delivered messages. When the application is finished with the message, the application simply calls Client.ack() on the message.

For messages that originated from a queue with at-least-once semantics, this adds the bookmark from the message to the batch of messages to acknowledge. For other messages, this method has no effect.

// Add the message to the next acknowledgment batch
client.ack(message);

To manually acknowledge processed messages and remove the messages from the queue, applications use the sow_delete command with the bookmarks of the messages to remove. Notice that AMPS only supports using a bookmark with sow_delete when removing messages from a queue, not when removing records from a SOW.

For example, given a Message object to acknowledge and a client, the code below acknowledges the message.

const acknowledgeSingle = async (client, message) => {
  return client.execute(
    new Command('sow_delete')
        .topic(message.header.topic())
        .bookmark(message.header.bookmark());
  );
};

In the example above, the program creates a sow_delete command, specifies the topic and the bookmark, and then sends the command to the server.

While this method works, creating and sending an acknowledgment for each individual message can be inefficient if your application is processing a large volume of messages. Rather than acknowledging each message individually, your application can build a comma-delimited list of bookmarks from the processed messages and acknowledge all of the messages at the same time. In this case, it's important to be sure that the number of messages you wait for is less than the maximum backlog -- the number of messages your client can have unacknowledged at a given time. Notice that both automatic acknowledgment and the convenience method Client.ack() take the maxiumum backlog into account.

A subscriber can also explicitly release a message back to the queue. AMPS returns the message to the queue, and redelivers the message just as though the lease had expired. To do this, the subscriber sends a sow_delete command with the bookmark of the message to release and the cancel option:

client.execute(
  new Command('sow_delete')
      .topic(message.header.topic())
      .bookmark(message.header.bookmark())
      .options('cancel');
);

To delta publish, you use the Client.deltaPublish() method as follows:

const data = ... ;  // obtain changed fields here
client.deltaPublish('my-topic', data);

The message that you provide to AMPS must include the fields that the topic uses to generate the SOW key. Otherwise, AMPS will not be able to identify the message to update. For SOW topics that use a User-Generated SOW Key, use the Command form of delta_publish to set the SowKey, as shown below:

const data = ... ;  // obtain changed fields here
const key = ... ;   // obtain user-generated SOW key

const command = new Command('delta_publish');
command.topic('delta-topic');
command.sowKey(key);
command.data(data);

/**
* Execute the delta publish. Do not provide a message
* handler since delta_publish is not waiting for the
* acknowledgment from the server.
*/
client.execute(command);

Delta messaging in AMPS has two independent aspects:

• Delta Subscribe - Allows subscribers to receive just the fields that are updated within a message.

• Delta Publish - Allows publishers to update and add fields within a message by publishing only the updates into the SOW.

This chapter describes how to create delta publish and delta subscribe commands using the AMPS JavaScript client. For a discussion of this capability, how it works, and how message types support this capability see the AMPS User Guide

AMPS queues are designed for high-volume applications that need minimal latency and overhead. One of the features that helps performance is the subscription backlog feature, which allows applications to receive multiple messages at a time. The subscription backlog sets the maximum number of unacknowledged messages that AMPS will provide to the subscription.

When the subscription backlog is larger than 1, AMPS delivers additional messages to a subscriber before the subscriber has acknowledged the first message received. This technique allows subscribers to process messages as fast as possible, without ever having to wait for messages to be delivered. The technique of providing a consistent flow of messages to the application is called smart pipelining.

Subscription Backlog

The AMPS server determines the backlog for each subscription. An application can set the maximum backlog that it is willing to accept with the max_backlog option. Depending on the configuration of the queue (or queues) specified in the subscription, AMPS may assign a smaller backlog to the subscription. If no max_backlog option is specified, AMPS uses a max_backlog of 1 for that subscription.

In general, applications that have a constant flow of messages perform better with a max_backlog setting higher than 1. The reason for this is that, with a backlog greater than 1, the application can always have a message waiting when the previous message is processed. Setting the optimum max_backlog is a matter of understanding the messaging pattern of your application and how quickly your application can process messages.

To request a max_backlog for a subscription, you explicitly set the option on the subscribe command, as shown below:

const command = new Command('subscribe')
                    .topic('my-queue')
                    .options('max_backlog=10');

To delta subscribe, you simply use the delta_subscribe command as follows:

const subId = await client.execute(

  // The delta_subscribe command to execute
  new Command('delta_subscribe')
      .topic('delta-topic')
      .filter('/thingIWant = "true"'),

  // Message handler
  message => { /* Delta messages arrive here */ }

);

// Once subscribed, it resolves with the subscription id.
console.log(subId);

The convenience method Client.deltaSubscribe() is available. If the delta_subscribe is not a valid command for the topic provided, a regular subscription will be created.

const subId = await client.deltaSubscribe(
  message => { /* ... */ },  // Message handler
  'delta-topic',             // Topic
  '/thingIWant = "true"'     // Filter (optional)
);

// Once subscribed, it resolves with the subscription id.
console.log(subId);

As described in the AMPS User Guide, messages provided to a delta subscription will contain the fields used to generate the SOW key and any changed fields in the message. Your application is responsible for choosing how to handle the changed fields.

The AMPS clients provide named convenience methods for core AMPS functionality. These named methods work by creating messages and sending those messages to AMPS. All communication with AMPS occurs through messages.

You can use the Command object to customize the messages that AMPS sends. This is useful for more advanced scenarios where you need precise control over the message, or in cases where you need to use an earlier version of the client to communicate with a more recent version of AMPS, or in cases where a named method is not available.

Understanding AMPS Messages and Commands

AMPS messages are represented in the client as Message objects. The Message object is generic, and can represent any type of AMPS message. This section includes a brief overview of elements common to Command - the class that is used to build AMPS messages. Full details of commands to AMPS are provided in the AMPS Command Reference Guide.

All AMPS Command objects contain the following elements:

• Command - The command tells AMPS how to interpret the message. Without a command, AMPS will reject the message. Examples of commands include publish, subscribe, and sow.

• CommandId - The command ID, together with the name of the client, uniquely identifies a command to AMPS. The command ID can be used later on to refer to the command or the results of the command. For example, the command ID for a subscribe command becomes the identifier for the subscription. The AMPS client provides a command ID when the command requires one and no command ID is set.

Most AMPS commands contain the following fields:

• Topic - The topic that the command applies to, or a regular expression that identifies a set of topics that the command applies to. For most commands, the topic is required. Commands such as logon, start_timer, and stop_timer do not apply to a specific topic, and do not need this field.

• Ack Type - The ack type tells AMPS how to acknowledge the message to the client. Each command has a default acknowledgment type that AMPS uses if no other type is provided.

• Options - The options are a comma-separated list of options that affect how AMPS processes and responds to the message.

Beyond these fields, different commands include fields that are relevant to that particular command. For example, SOW queries, subscriptions, and some forms of SOW deletes accept the Filter field, which specifies the filter to apply to the subscription or query. As another example, publish commands accept the Expiration field, which sets the SOW expiration for the message.

For full details on the options available for each command and the acknowledgment messages returned by AMPS, see the AMPS Command Reference Guide.

Creating and Populating a Command

To create a command, you simply construct a command object of the appropriate type:

var command = new amps.Command('sow');

Once created, you set the appropriate fields on the command. For example, the following code creates a publish message, setting the command, topic, data to publish, and an expiration for the message:

var command = new amps.command('publish')
                  .topic('messages')
                  .data({id: 1, hello: 'World'})
                  .expiration(5);

Another example is a SOW command:

var command = new amps.command('sow')
                  .topic('messages-sow')
                  .filter('/id > 20');

When sent to AMPS using the Client.execute() method, AMPS performs a SOW query from the topic messages-sow using a filter of /id > 20. The results of sending this message to AMPS are no different than using the form of the sow method that sets these fields.

Using Execute

Once you've created a command, use the Client.execute() method to send the command to AMPS. The method returns the Promise object, sends the message to AMPS, waits for a processed acknowledgment, then resolves the promise with the command ID. Messages are processed on the client asynchronously.

For example, the following snippet sends the command created above:

client
    .execute(command, messageHandler)
    .then(function(commandId) { ... })
    .catch(...);

This is equivalent to calling the Client.sow() method.

You can also provide a message handler to receive acknowledgments, statistics, or the results of subscriptions and SOW queries:

function messageHandler(message) {
    // Received the requested ack 'processed' message
    if (message.c === 'ack') {
        console.log(message.a + ':', message.reason);
    }
    else {
        // other messages are handled here
        // ...
    }
}

client.execute(
    // request the ack message from AMPS
    command.ackType('processed'),

    // Message handler
    messageHandler
);

While this message handler simply prints the ack type and reason for sample purposes, message handlers in production applications are typically designed with a specific purpose. For example, your message handler may fill a work queue, or check for success and throw an exception if the command failed.

Notice that the publish command does not typically return results other than acknowledgment messages. To send a publish command, use the execute() method, without providing a message handler:

client.execute(publishCmd);

Transport Filtering

The AMPS JavaScript client offers the ability to filter incoming and outgoing messages in the format they are sent and received on the network. This allows you to inspect or modify outgoing messages before they are sent to the network, and incoming messages as they arrive from the network. This can be especially useful when using SSL connections, since this gives you a way to monitor outgoing network traffic before it is encrypted, and incoming network traffic after it is decrypted.

To create a transport filter, you create a callable that expects a string that contains the raw data, and a direction parameter indicating whether the string is output or not. For example, the following function simply prints the direction and data:

const printingFilter = (data, outgoing) => {
  if (outgoing) {
    console.log('OUTGOING ---> ', data);
  }
  else {
    console.log('INCOMING ---> ', data.data);
  }
};

You then register the filter by calling Client.transportFilter() with the function, as shown below.

// client is an AMPS client
client.transportFilter(printingFilter);

Using WebSocket over SSL

The AMPS JavaScript client includes support for Secure Sockets Layer. To use this support in the JavaScript client, you need only use wss for the transport type in the connection string.

Imagine an application that displays real time information about the position and status of a fleet of delivery vans. When the application starts, it should display the current location of each of the vans along with their current status. As vans move around the city and post other status updates, the application should keep its display up to date. Vans upload information to the system by posting messages to the van_location topic, configured with a key of van_id on the AMPS server.

In this application, it is important to not only stay up-to-date on the latest information about each van, but to ensure all of the active vans are displayed as soon as the application starts. Combining a SOW with a subscription to the topic is exactly what is needed, and that is accomplished by the Client.sowAndSubscribe() method, or by executing a sow_and_subscribe command.

sowAndSubscribe()

First, let's look at an example that uses the convenience method:

const reportVanPosition = async client => {
  /**
  * sowAndSubscribe() method to begin receiving information about
  * all of the active delivery vans in the system. All of the vans
  * in the system now are returned as Message objects whose
  * `message.header.command()` method returns `sow`. New messages
  * coming in are returned as Message objects whose
  * `message.header.command()` method returns `p` (publish).
  */
  return client.sowAndSubscribe(

    // Message handler
    message => {
      const cmdName = message.header.command();
      if (cmdName === 'sow' || cmdName === 'p') {
        /**
        * For each of these messages we call addOrUpdateVan(),
        * that presumably adds the van to our application’s
        * display. As vans send updates to the AMPS server,
        * those are also received by the client because of the
        * subscription placed by sowAndSubscribe(). Our
        * application does not need to distinguish between
        * updates and the original set of vans we found via the
        * SOW query, so we use addOrUpdateVan() to display
        * the new position of vans as well.
        */
        addOrUpdateVan(message);
      }
      else if (cmdName === 'oof') {
        removeVan(message);
      }
    },

    'van_location',         // SOW Topic

    '/status = "ACTIVE"',   // Filter

    // Additional parameters
    {
      batchSize: 100,
      options: 'oof'
    }
  );
};

Execute a Command

Now we will look at an example that uses the Command interface with the Client.execute() method:

// Message Handler
const onVanPositionUpdate = message => {
  const cmdName = message.header.command();
  if (cmdName === 'sow' || cmdName === 'p') {
    addOrUpdateVan(message);
  }
  else if (cmdName === 'oof') {
    removeVan(message);
  }
};

const reportVanPosition = async client => {
  // Command object to execute
  const cmd = new Command('sow_and_subscribe');
  cmd.topic('van_location');
  cmd.filter('/status = "ACTIVE"');
  cmd.batchSize(100);
  cmd.options('oof');

  // Execute the command with the above message handler
  return client.execute(cmd, onVanPositionUpdate);
};

Notice that the two forms have the same result.

OOF Messages

In the above examples we specified the oof option to the command. Setting this option causes AMPS to send Out-of-Focus (OOF) messages for the topic. OOF messages are sent when an entry that was sent to us in the past no longer matches our query. This happens when an entry is removed from the SOW cache via a sow_delete operation, when the entry expires (as specified by the expiration time on the message or by the configuration of that topic on the AMPS server), or when the entry no longer matches the content filter specified. In our case, if a van's status changes to something other than ACTIVE, it no longer matches the content filter, and becomes out of focus. When this occurs, a message is sent with Command set to oof. We use OOF messages to remove vans from the display as they become inactive, expire, or are deleted.

The AMPS JavaScript Client provides an easy way to create highly-available applications using AMPS, via the Client class. In other clients the highly available version of the client is called HAClient, however in JavaScript the Client class has all the functions of Client, providing both regular and highly available modes. Among its regular functions, Client provides protection against network, server, and client outages.

Using High Availability (HA) functions of the Client allows applications to automatically:

• Recover from temporary disconnects between client and server.

• Failover from one server to another when a server becomes unavailable.

Since the Client can automatically manage failover and reconnection, 60East recommends using the HA functions for applications that need to:

• Ensure no messages are lost or duplicated after a reconnect or failover.

• Persist messages and bookmarks in memory for protection against client failure.

You can choose how your application uses HA features. For example, you might need automatic reconnection, but have no need to resume subscriptions or republish messages. The high availability behavior in Client is provided by implementations of defined interfaces. You can combine different implementations provided by 60East to meet your needs, and implement those interfaces to provide your own policies.

Some of these features require specific configuration settings on your AMPS instance(s). This chapter mentions these features and describes how to use them from the AMPS JavaScript client. You can find full documentation for these settings and server features in the AMPS User Guide.

Reconnection with Client

This description provides a high-level framework for understanding the components involved in failover with the Client. The components are described in more detail in the following sections.

The Client reconnect handler performs the following steps when reconnecting:

1. Calls the ServerChooser to determine the next URI to connect to and the Authenticator to use for that connection.

   If the connection fails, calls getError() on the ServerChooser to get a description of the failure, sends an error to the error handler, and stops the reconnection process.

2. Calls the DelayStrategy to determine how long to wait before attempting to reconnect, and waits for that period of time.

3. Connects to the AMPS server. If the connection fails, calls reportFailure() on the ServerChooser and begins the process again.

4. Logs on to the AMPS server. If the connection fails, calls reportFailure() on the ServerChooser and begins the process again.

5. Calls reportSuccess() on the ServerChooser.

6. Receives the bookmark for the last message that the server has persisted. Discards any older messages from the PublishStore.

7. Republishes any messages in the PublishStore that have not been persisted by the server.

8. Re-establishes subscriptions using the SubscriptionManager for the client. For bookmark subscriptions, the reconnect handler uses the BookmarkStore for the client to determine the most recent bookmark, and resubscribes with that bookmark. For subscriptions that do not use a bookmark, the SubscriptionManager simply re-enters the subscription, meaning that it is entered at the point at which the Client reconnects.

The ServerChooser, DelayStrategy, PublishStore, SubscriptionManager, and BookmarkStore are all extension points for the Client. You can adapt the failover and recovery behavior by setting a different object for the behavior you want to customize on the Client or by providing your own implementation.

Choosing Store Durability

The Client provides recovery after disconnection using Stores. As the name implies, stores hold information about the state of the client.

These stores provide the following capabilities:

• A bookmark store tracks received messages, and is used to resume subscriptions.

• A publish store tracks published messages, and is used to ensure that messages are persisted in AMPS.

The AMPS JavaScript client provides a memory-backed version of each store. The store interface is public, and an application can create and provide a custom store as necessary. A Client can use a memory backed store for protection, as described below:

• Memory-backed stores provide disconnection recovery from AMPS by storing messages and bookmarks in your process' address space. This is the highest performance option for working with AMPS in a highly available manner. The trade-off with this method is there is no protection from a crash or failure of your client application. If your application is terminated prematurely or, if the application terminates at the same time as an AMPS instance failure or network outage, then messages may be lost or duplicated.

  A memory-backed store should only be used by one instance of a client at a time.

The store interface is public, and an application can create and provide a custom store as necessary. While clients provide convenience methods for creating memory-backed Client objects with the appropriate stores, you can also create and set the stores in your application code.

The Client provides convenience methods for creating clients and setting stores. You can also construct a Client and set the store implementations you choose.

In this example, we create several clients. The first client has the full set of HA features, such as: supports automatic connection failover and reconnection, uses memory stores for both bookmarks and publishes, and automatically re-subscribes in case of a failover. The second client does not set a store for publishes, which means that AMPS will not store outgoing messages, but is using the bookmark store to track incoming messages. The final client does not specify any stores, and so has no persistence for published messages or bookmark subscriptions, but takes advantage of the automatic failover and reconnection in the Client.

// Failover, Bookmark and Publish stores, re-subscription
const memoryClient = Client.createMemoryBacked('memory-backed');

// No publish store, no failover, memory-backed bookmark store
const subscriberClient = new Client('subscriber');
subscriberClient.bookmarkStore(new MemoryBookmarkStore());

// Failover behavior only.
const failoverClient = new Client('failover');
const serverChooser = new DefaultServerChooser();
serverChooser.add('ws://localhost:9000/amps/json');
serverChooser.add('ws://localhost:9100/amps/json');
serverChooser.add('ws://localhost:9200/amps/json');

failoverClient.serverChooser(serverChooser);
failoverClient.delayStrategy(new FixedDelayStrategy());

Connections and the Server Chooser

In the high availability mode, Client attempts to keep itself connected to an AMPS instance at all times, by automatically reconnecting or failing over when it detects that the client is disconnected. When you are using the Client directly, your disconnect handler usually takes care of reconnection. Client with HA classes on the other hand, can provide a disconnect handler that automatically reconnects to the current server or to the next available server.

To inform the Client of the addresses of the AMPS instances in your system, you pass a ServerChooser instance to the Client. ServerChooser acts as a smart enumerator over the servers available: Client calls ServerChooser methods to inquire about what server should be connected, and calls methods to indicate whether a given server succeeded or failed.

The AMPS JavaScript Client provides a simple implementation of ServerChooser, called DefaultServerChooser, that provides very simple logic for reconnecting. This server chooser is most suitable for basic testing, or in cases where an application should simply rotate through a list of servers. For most applications, you implement the ServerChooser interface yourself for more advanced logic, such as choosing a backup server based on your network topology, or limiting the number of times your application should try to reconnect to a given address.

To connect to AMPS, you provide a ServerChooser to Client and then call Client.connect() to create the first connection:

const memoryClient = new Client('server-chooser-demo');

// primary.amps.xyz.com is the primary AMPS instance, and
// secondary.amps.xyz.com is the secondary
const chooser = new DefaultServerChooser();

chooser.add('ws://primary.amps.xyz.com:12345/amps/fix');
chooser.add('ws://secondary.amps.xyz.com:12345/amps/fix');

memoryClient.serverChooser(chooser);

try {
  await memoryClient.connect();

  // ... later
  await client.disconnect();
}
catch (err) {
  // handle connection error
}

Client remains connected to the server until disconnect() is called. Client automatically attempts to reconnect to your server if it detects a disconnect and, if that server cannot be connected, fails over to the next server provided by the ServerChooser. In this example, the call to connect() attempts to connect and login to primary.amps.xyz.com, and resolves if that is successful. If it cannot connect, it tries secondary.amps.xyz.com, and continues trying servers from the ServerChooser until a connection is established. Likewise, if it detects a disconnection while the client is in use, then Client attempts to reconnect to the server with which it was most recently connected; if that is not possible, then it moves on to the next server provided by the ServerChooser.

The default ServerChooser simply provides the next URL in the sequence. This strategy works for many applications. If you need a different strategy, you can implement your own logic for failover by creating a class derived from ServerChooser.

Setting a Reconnect Delay and Timeout

You can control the amount of time between reconnection attempts and set a total amount of time for the Client to attempt to reconnect.

The AMPS JavaScript client includes a method for setting a delay strategy on a client, Client.delayStrategy(). This method accepts an instance of any type that provides the methods getConnectWaitDuration() and reset(), as described in the API documentation.

While you can easily implement your own delay strategy, the client also provides two delay strategies:

• FixedDelayStrategy provides the same delay each time the Client tries to reconnect.

• ExponentialDelayStrategy provides an exponential backoff until a connection attempt succeeds.

To use either of these classes, you simply create an instance, set the appropriate parameters, and install that instance as the delay strategy for the Client. For example, the following code sets up a reconnect delay that starts at 200ms and increases the delay by 1.5 times after each failure. The strategy allows a maximum delay between connection attempts of 5 seconds, and will not retry longer than 60 seconds.

const client = new Client('delay-strategy-demo');

client.delayStrategy(
  new ExponentialDelayStrategy({
    initialDelay: 200,
    maximumDelay: 5 * 1000,
    backoffExponent: 1.5,
    maximumRetryTime: 60 * 1000
  })
);

Implementing a Server Chooser

As described above, you provide the Client with connection strings to one or more AMPS servers using a ServerChooser. The purpose of a ServerChooser is to provide information to the Client. A ServerChooser does not manage the reconnection process, and should not call methods on the Client.

A ServerChooser has two required responsibilities to the Client:

• Tells the Client the connection string for the server to connect to. If there are no servers, or the ServerChooser wants the connection to fail, the ServerChooser returns null.

  To provide this information, the ServerChooser implements the getCurrentUri() method.

• Provides an Authenticator for the current connection string. This is especially important for installations where different servers require different credentials or authentication tokens must be reset after each connection attempt.

  To provide the authenticator, the ServerChooser implements the getCurrentAuthenticator() method.

The Client calls the getCurrentUri() and getCurrentAuthenticator() methods each time it needs to make a connection.

Each time a connection succeeds, the Client calls the reportSuccess() method of the ServerChooser. Each time a connection fails, the Client calls the reportFailure() method of the ServerChooser. The Client does not require the ServerChooser to take any particular action when it calls these methods. These methods are provided for the Client to do internal maintenance, logging, or record keeping. For example, a Client might keep a list of available URIs with a current failure count, and skip over URIs that have failed more than 5 consecutive times until all URIs in the list have failed more than 5 consecutive times.

When the ServerChooser returns a null from getCurrentUri(), indicating that no servers are available for connection, the Client calls the getError() method on the ServerChooser, if one is provided, and includes the string returned by getError() in the generated exception.

Heartbeats and Failure Detection

Use of the Client allows your application to quickly recover from detected connection failures. By default, connection failure detection occurs when AMPS receives an operating system error on the connection. This system may result in unpredictable delays in detecting a connection failure on the client, particularly when failures in network routing hardware occur, and the client primarily acts as a subscriber.

The heartbeat feature of the AMPS client allows connection failure to be detected quickly. Heartbeats ensure that regular messages are sent between the AMPS client and server on a predictable schedule. The AMPS client and server both assume disconnection has occurred if these regular heartbeats cease, ensuring disconnection is detected in a timely manner. To use the heartbeat feature, call the heartbeat() method on Client:

const memoryClient = new Client('importantStuff');

// ...

memoryClient.heartbeat(3);
await memoryClient.connect();

Method heartbeat() takes one parameter: the heartbeat interval. The heartbeat interval specifies the periodicity of heartbeat messages sent by the server: the value 3 indicates messages are sent on a three-second interval. If the client receives no messages in a six-second window (two heartbeat intervals), the connection is assumed to be dead, and the Client attempts reconnection. The optional second parameter of the heartbeat() method allows the idle period to be set to a value other than two heartbeat intervals.

Considerations for Publishers

Publishing with the Client in the HA mode is nearly identical to regular publishing; you simply call the publish() method with your message’s topic and data. The AMPS client sends the message to AMPS, and then returns from the publish() call. For maximum performance, the client does not wait for the AMPS server to acknowledge that the message has been received.

When a Client sets a publish store, the publish store retains a copy of each outgoing message and requests that AMPS acknowledge that the message has been persisted. The AMPS server acknowledges messages back to the publisher. Acknowledgments can be delivered for multiple messages at periodic intervals (for topics recorded in the transaction log) or after each message (for topics that are not recorded in the transaction log). When an acknowledgment for a message is received, the Client removes that message from the publish store. When a connection to a server is made, the Client automatically determines which messages from the publish store (if any) the server has not processed, and replays those messages to the server once the connection is established.

For reliable publishers, the application must choose how best to handle application shutdown. For example, it is possible for the network to fail immediately after the publisher sends the message, while the message is still in transit. In this case, the publisher has sent the message, but the server has not processed it and acknowledged it. During normal operation, the Client will automatically connect and retry the message. On shutdown, however, the application must decide whether to wait for messages to be acknowledged, or whether to exit.

Publish store implementations provide an unpersistedCount() method that reports the number of messages that have not yet been acknowledged by the AMPS server. When the unpersistedCount() reaches 0, there are no unpersisted messages in the local publish store.

For the highest level of safety, an application can wait until the unpersistedCount() reaches 0, which indicates that all of the messages have been persisted to the instance that the application is connected to, and the synchronous replication destinations configured for that instance. When a synchronous replication destination goes offline, this approach will cause the publisher to wait to exit until the destination comes back online or until the destination is downgraded to asynchronous replication.

For applications that are shut down periodically for short periods of time (for example, applications that are only offline during a weekly maintenance window), another approach is to use the Client.flush() method to ensure that messages are delivered to AMPS, and then rely on the connection logic to replay messages as necessary when the application restarts.

For example, the following code flushes messages to AMPS, then warns if not all messages have been acknowledged:

const client = Client.createMemoryBacked('ha-publisher');
// ...
await client.connect();

// Publish messages
client.publish('topic', {id: 1});
client.publish('topic', {id: 2});
// ...

/**
* We think we are done, but the server may not
* have received or acknowledged all messages yet.
* Wait until the server has received all messages.
* The program could also specify a timeout in this
* command to avoid blocking forever if the network
* is down or all servers are offline.
*/
await client.flush();

/**
* Print warning to the console if messages have
* been published but not yet acknowledged as persisted
*/
if (client.publishStore().unpersistedCount() > 0) {
  console.log('All messages have been published');
  console.log('But not all have been persisted');
}

client.disconnect();

In this example, the client sends each message immediately when publish() is called. If AMPS becomes unavailable between the final publish() and the disconnect(), or one of the servers that the AMPS instance replicates to is offline, the client may not have received a persisted acknowledgment for all of the published messages. For example, if a message has not yet been persisted by all of the servers in the replication fabric that are connected with synchronous replication, AMPS will not have acknowledged the message.

Before shutting down the client, the code does two things:

• First, the code flushes messages to the server to ensure that all messages have been delivered to AMPS.

• Next, the code checks to see if all of the messages in the publish store have been acknowledged as persisted by AMPS. If the messages have not been acknowledged, they will remain in the publish store and will be published to AMPS, if necessary, the next time the client connects. An application may choose to wait until unpersistedCount() returns 0, or (as we do in this case) simply warn that AMPS has not confirmed that the messages are fully persisted. The behavior you choose in your application should be consistent with the high-availability guarantees your application needs to provide.

Considerations for Subscribers

Client provides two important features for applications that subscribe to one or more topics: re-subscription, and a bookmark store to track the correct point at which to resume a bookmark subscription.

Resubscription With SubscriptionManager

Any asynchronous subscription placed using a Client is automatically reinstated after a disconnect or a failover. These subscriptions are placed in an in-memory SubscriptionManager, which is created automatically when the Client.createMemoryBacked() static method is called. Alternatively, it can be created and assigned before the client is connected:

const client = new Client('subscription-manager-demo');
client.subscriptionManager(new DefaultSubscriptionManager());

Most applications will use this built-in subscription manager, but for applications that create a varying number of subscriptions, you may wish to implement SubscriptionManager to store subscriptions in a more durable place. Note that these subscriptions contain no message data, but rather simply contain the parameters of the subscription itself (for instance, the command, topic, message handler, options, and filter).

When a re-subscription occurs, the AMPS JavaScript Client re-executes the command as originally submitted, including the original topic, options, and so on. AMPS sends the subscriber any messages for the specified topic (or topic expression) that are published after the subscription is placed. For a sow_and_subscribe command, this means that the client reissues the full command, including the SOW query as well as the subscription.

Bookmark Stores

In cases where it is critical not to miss a single message, it is important to be able to resume a subscription at the exact point that a failure occurred. In this case, simply recreating a subscription isn't sufficient. Even though the subscription is recreated, the subscriber may have been disconnected at precisely the wrong time, and will not see the message.

To ensure delivery of every message from a topic or set of topics, the AMPS Client can plug in a BookmarkStore that, combined with the bookmark subscription and transaction log functionality in the AMPS server, ensures that clients receive any messages that might have been missed. The client stores the bookmark associated with each message received, and tracks whether the application has processed that message; if a disconnect occurs, the client uses the BookmarkStore to determine the correct resubscription point, and sends that bookmark to AMPS when it re-subscribes. AMPS then replays messages from its transaction log from the point after the specified bookmark, thus ensuring the client is completely up-to-date.

Client helps you to take advantage of this bookmark mechanism through the Client.bookmarkStore() method and MemoryBookmarkStore class. When a bookmark store is assigned to the client, whenever a disconnection or failover occurs, your application automatically re-subscribes to the message after the last message it processed.

To take advantage of bookmark subscriptions, do the following:

• Ensure the topic(s) to be subscribed to are included in a transaction log. See the AMPS User Guide for information on how to specify the contents of a transaction log.

• Before connecting, create and assign a bookmark store object to the client.

• Use the Client.bookmarkStore().discard() method in message handlers to indicate when a message has been fully processed by the application.

The following example creates a bookmark subscription against a transaction-logged topic, and fully processes each message as soon as it is delivered:

const client = new Client('aClient');
client.subscriptionManager(new DefaultSubscriptionManager());
client.bookmarkStore(new MemoryBookmarkStore());

// ...

await client.execute(
 // The subscription command
 new Command('subscribe')
     .topic('myTopic')
     .bookmark(Client.Bookmarks.MOST_RECENT)
     .subId('MySubId'),

 // Message handler discards every message after processing
 message => {
   console.log(message.data);
   client.bookmarkStore().discard(message);
 }
);

Storing these bookmarks in the bookmark store allows the application to restart the subscription from the last message processed, in the event of either server failure or disconnect.

The command method, subId(), specifies an identifier to be used for this subscription. If the subId is not provided, Client will generate one and resolve the Client.execute() Promise with it, like most other Client functions. If you wish to resume a subscription from a previous point after the application has disconnected, the application must pass the same subscription ID as before. Passing a different subscription ID bypasses any recovery mechanisms, creating an entirely new subscription. When you use an existing subscription ID, the Client locates the last-used bookmark for that subscription in the bookmark store, and attempts to re-subscribe from that point.

• Client.Bookmarks.NOW specifies that the subscription should begin from the moment the server receives the subscription request. This results in the same messages being delivered as if you had invoked subscribe() instead, except that the messages will be accompanied by bookmarks. This is also the behavior that results if you supply an invalid bookmark.

• Client.Bookmarks.EPOCH specifies that the subscription should begin from the beginning of the AMPS transaction log (that is, the first entry in the oldest journal file for the transaction log).

• Client.Bookmarks.MOST_RECENT specifies that the subscription should begin from the last-used message in the associated BookmarkStore. Alternatively, if this subscription has not been seen before, it instructs the subscription to begin with EPOCH. This is the most common value for this parameter, and is the value used in the preceding example. By using MOST_RECENT, the application automatically resumes from wherever the subscription left off, taking into account any messages that have already been processed and discarded.

When the Client re-subscribes after a disconnection and reconnection, it always uses MOST_RECENT, ensuring that the continued subscription always begins from the last message used before the disconnect, so that no messages are missed.

Conclusion

With only a few changes, most AMPS applications can take advantage of the high availability features of the Client to become more highly-available and resilient. Using the PublishStore, publishers can ensure that every message published has actually been persisted by AMPS. Using BookmarkStore, subscribers can make sure that there are no gaps or duplicates in the messages received. Client makes both kinds of applications more resilient to network and server outages and temporary issues. Though Client provides useful defaults for the PublishStore, BookmarkStore, SubscriptionManager, ServerChooser, and DelayStrategy, you can customize any or all of these to the specific needs of your application and architecture.

Unlike other AMPS client libraries, the JavaScript library parses message data contents before delivering the message to a handler. This happens for all the supported message types.

By default, the client supports the following message types:

• JSON;

• FIX / NVFIX - via FixTypeHelper class;

• Binary.

Messages of these types are automatically parsed into native JavaScript data structures and are ready for consumption.

The client library utilizes the concept of a type helper for message data processing. It’s an inteface with a set of methods for parsing and serializing message data of a paricular type.

The provided static TypeHelper class is used for fine grained control over message types. It is used to register new composite and custom message types, or replace type helpers for existing types.

Composite Message Types

The TypeHelper.compositeHelper static method handles creating and parsing composite message types:

// Register the json-xml-json-binary composite message type
TypeHelper.helper(
  // The name of the new composite message type
  'compositejxjb',

  // create the composite type helper
  TypeHelper.compositeHelper(
    'json',    // Part 1: JSON
    'xml',     // Part 2: XML
    'json',    // Part 3: JSON
    'binary'   // Part 4: Binary
  )
);

The composite type helper created with TypeHelper.compositeHelper will automatically build data to send from the array of the message parts and parse multi-part messages from the received raw data. All types used in the new composite type helper should be registered before creating it.

Building Composite Messages

Once the composite type helper was registered, it becomes very easy to build composite messages of that type. In the following example, we register a new composite message type, compositejjjb that consists of three JSON parts and one binary part. Then, we create the composite message and publish it. The type helper will take care of converting message parts into the raw message data that will be sent:

// Before creating a client, register the composite message type
TypeHelper.helper(
  'compositejjjb',
  TypeHelper.compositeHelper('json', 'json', 'json', 'binary')
);

const client = new Client('composite-sender');
await client.connect('ws://localhost:9000/amps/compositejjjb');

// create the array with message parts
const parts = [
  {id: 5, value: 'part1'},        // JSON part 1
  {value: 'part2', data: 22.22},  // JSON part 2
  {value: 'part3', data: 33.33},  // JSON part 3
  'XXXXXXXXXXXXXXXXXXXXXXXXXXX'   // Binary data
];

// Publish the composite message
client.publish('composite-topic', parts);

Parsing Composite Messages

Once the composite type helper is registered, the composite messages of that type will be parsed automatically. The Message.data field will contain the array of message parts, parsed according to their types:

await client.subscribe(
  message => {
    const parts = message.data;
    console.log('Received message with', parts.length, 'parts');
    parts.forEach(part => console.log(part));
  },
  'composite-topic'
);

Notice that the receiving application is written with explicit knowledge of the structure and content of the composite message type.

Custom Message Types

If a message type used in your application is not supported by default, it is possible to create a new type helper for it. Another situation in which you might need a custom type helper is when the default implementation does not fit your needs. For example, the default implementation of FIX / NVFIX message types assumes that the keys in each message are unique, thus overriding values if the same key occurs twice in the same message. If your messages contain duplicated keys and that is expected behavior, you need to override the default type helper with a custom implementation.

Each type helper is an object that must contain the following methods:

• serialize(data: any): string[] - this method is used to serialize data in order send it to the server. All data chunks should be converted into an array of strings.

• deserialize(data: any): any - this method deserializes data from the Server into a format that will be consumed by the message handler. It can be any format as long as the consumer of the data is aware of it.

Below we provide some examples on how to utilize TypeHelper functionality.

Create New Type Helpers

Create and register a custom type helper for XML messages:

class XMLTypeHelper {
  serialize(data) {
    return [new XMLSerializer().serializeToString(data)];
  }

  deserialize(data) {
    if (data.constructor === String) {
      return new DOMParser().parseFromString(data);
    }

    // Binary buffer, need to decode utf8
    return new DOMParser().parseFromString(
      decodeURIComponent(escape(Uint8ToString(data)))
    );
  }
}

// Register the XML helper for parsing XML messages
TypeHelper.helper('xml', new XMLTypeHelper());

Override Default Type Helpers

In case the custom parsing behavior is expected, it is possible to override the default type helper:

/**
* create a JSON type helper that does not parse JSON data into native
* JS objects keeping it in form of a string
*/
const jsonHelper = {
  serialize: data => [data],
  deserialize: data => JSON.stringify(data)
};

// override the default type helper
TypeHelper.helper('json', jsonHelper);

More examples are provided in the JavaScript client library API Documentation.

FIX / NVFIX delimiters

In some cases the delimiter value used in FIX / NVFIX messages differs from the default (\x01). In this situation you don’t have to implement a custom type helper; instead, the custom delimiter can be set:

TypeHelper.helper('nvfix').delimiter('%01');