SOW and Subscribe

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 also 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 AMPS sow_and_subscribe command. Now we will look at an example:

private void updateVanPosition(Message message) {
    switch (message.getCommand()) {
        case Message.Command.SOW:
        case Message.Command.Publish:
            /* For each of these messages, addOrUpdateVan() 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);
            break;
        case Message.Command.OOF:
            removeVan(message);
        break;
    }
}

public void subscribeToVanLocation(Client client) {
    try {
        Command command = new Command("sow_and_subscribe")
                            .setTopic("van_location")
                            .setFilter("/status = 'ACTIVE'")
                            .setBatchSize(100)
                            .setOptions("oof");


        /* We issue a sowAndSubscribe() 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 Messages whose getCommand() returns SOW.
         */
        for (Message message : client.execute(command)) {
                updateVanPosition(message);
        }
    }
    catch (AMPSException aex) {
        System.err.println("TestListener caught exception.");
    }
}

public void addOrUpdateVan(message) {

    // Use information in the message to add the van or update
    // the van position.

    ...

}

public void removeVan(message) {

   // Use information in the message to remove information on
   // the van position.

   ...
}

Now we will look at an example that uses the asynchronous form of sowAndSubscribe:

public class VanPositionUpdater {

    public void invoke(Message message) {
        updateVanPosition(message);
    }

    private void updateVanPosition(Message message) {
        switch (message.getCommand()) {
            case Message.Command.SOW:
            case Message.Command.Publish:
                addOrUpdateVan(message);
                break;
            case Message.Command.OOF:
                removeVan(message);
                break;
        }
    }

    public void addOrUpdateVan(message) {
       // Use information in the message to add the van or update
       // the van position.
       ...
     }

     public void removeVan(message) {
      // Use information in the message to remove information on
      // the van position.
      ...
     }
}


public void subscribeToVanLocation(Client client) {
    try {

        VanPositionUpdater vp = new VanPositionUpdater();
        Command command = new Command("sow_and_subscribe")
                            .setTopic("van_location")
                            .setFilter("/status = 'ACTIVE'")
                            .setBatchSize(100)
                            .setOptions("oof_enabled");

        client.execute(command, vp);

    }
    catch (AMPSException aex) {
        System.err.println("TestListener caught exception.");
    }
}