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."
}

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.

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
);

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.

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.

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.

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.