In Chapter 4, Implementing Domain Objects, we implemented the domain model for our sample application. One of the classes we created is the Account class, which has a method that allows us to withdraw the money from the account:

public Money withdraw(Money amount, String description)
throws InsufficientFundsException {
Money balance = m_balance;
// currency conversion code omitted for brevity
if (amount.greaterThan(balance)) {
throw new InsufficientFundsException(balance, amount);
}
m_balance = balance = balance.subtract(amount);
postTransaction(TransactionType.WITHDRAWAL,
description,
amount,
balance);
return balance;
}

As highlighted above, this method mutates the internal state of the Account instance, so we need to provide some kind of concurrency control. Using explicit locking is an option, but for the reasons discussed earlier we will create and use a custom entry processor instead:

public class WithdrawalProcessor
extends AbstractProcessor
implements Serializable {
private Money m_amount;
private String m_description;
public WithdrawalProcessor(Money amount, String description) {
m_amount = amount;
m_description = description;
}
public Object process(InvocableMap.Entry entry) {
Account account = (Account) entry.getValue();
Money balance = account.withdraw(amount, description);
entry.setValue(account, false);
return balance;
}
}

It might be obvious by now that entry processors are my favorite way to perform any mutating operations on cached objects, but the example above also shows my favorite way to invoke such operations. All the business logic is completely encapsulated within domain classes, and the entry processor is simply used to control the location of execution.

I have also seen people implement business logic within the entry processor, but I don't personally like that approach. I don't believe that a Coherence-specific infrastructure component, such as the entry processor, is the best place to put business logic.

We have discussed data affinity, so we know that the new transaction should be stored on the same node (and within the same partition) as the account it is related to. Because the entry processor by definition executes on that same node, we can access the backing map for the transactions cache directly and put the transaction into it.

We can easily achieve this by overriding the save method within the CoherenceTransactionRepository implementation:

public void save(Transaction tx) {
CacheService service = getCache().getCacheService();
BackingMapManager bmm = service.getBackingMapManager();
BackingMapManagerContext ctx = bmm.getContext();
Converter keyConverter = ctx.getKeyToInternalConverter();
Converter valueConverter = ctx.getValueToInternalConverter();
Map backingMap = ctx.getBackingMap(getCache().getCacheName());
backingMap.put(keyConverter.convert(tx.getId()),
valueConverter.convert(tx));
}

As you can see, accessing the backing map directly is significantly more complex than accessing named cache, but it is not rocket science. You first need to obtain a BackingMapManager from the cache service and get its context. The backing map stores entries in an internal (binary) format, so you also need converters for both the key and the value, which can be obtained from the BackingMapManagerContext. Finally, you need to get the backing map itself and put the converted entry into it.

This effectively bypasses the cache service completely and avoids re-entrancy issues. However, keep in mind that direct backing map access is a very advanced feature and don't use it unless you know exactly what you are doing and are absolutely certain that no simpler solution is available.

These two processors allow you to implement optimistic concurrency for your objects.

Unlike the NamedCache.put method, which will simply overwrite the existing value with the new value even if it has been changed in the meantime, the VersionedPut method will update the entry only if the versions of the existing and the new value match.

In order to be able to use one of these processors, your objects need to implement a com.tangosol.util.Versionable interface, which defines the following methods:

public interface Versionable {
public Comparable getVersionIndicator();
public void incrementVersion();
}

By default, VersionedPut will not return anything and will only update an entry if it is already in the cache. However, you can change this behavior to return the existing value if the versions do not match, or to insert new value into the cache if it doesn't already exist.

The PriorityProcessor allows you to explicitly control the scheduling priority and execution timeout of any entry processor.

This is an advanced feature that should be used judiciously. It can come in very handy in situations where you need to ensure that the processor is executed as soon as possible, or to override a default timeout value for a long-running processor.

The ExtractorProcessor allows you to extract a value from a cache entry using a ValueExtractor.

This processor is important because it can use available indexes to avoid deserialization of a value, which can significantly improve performance.

The UpdaterProcessor uses a ValueUpdater to modify one or more properties of a target object without sending the whole object across the wire.

This can be very useful when you need to update only a few properties on a large object, as it can significantly reduce the amount of network traffic. It essentially allows you to send only a small delta containing changes and update the target object on the node that owns it.

You can define the task that you want to execute in parallel by implementing the Work interface. For the most part, this boils down to implementation of the run method defined by the Runnable interface, which Work extends. However, there are two more methods you need to implement: isDaemon and release.

The isDaemon method is ignored by the Coherence implementation of Work Manager, so you can simply return false from it. The release method is called indirectly when you invoke the method with the same name on the related WorkItem instance. If your run method is iterative and could potentially execute for a long time, you should use the release method to set a flag that can be used to interrupt the iteration and perform necessary cleanup.

That said, the run method is where you will implement the custom logic your task should execute. For example, if you need to send e-mail alerts to customers when their account balance falls below a predefined amount, you might want to implement a task for that purpose:

public class LowBalanceAlertTask
implements Work, PortableObject {
private Account account;
public LowBalanceAlertTask() {
// deserialization constructor
}
public LowBalanceAlertTask(Account account) {
this.account = account;
}
public void release() {
}
public boolean isDaemon() {
return false;
}
public void run() {
SimpleMailMessage message = new SimpleMailMessage();
message.setFrom("[email protected]");
message.setTo(account.getCustomer().getEmailAddress());
message.setSubject("ALERT: Low Balance");
message.setText("Your account balance is " + account.getBalance());
JavaMailSenderImpl sender = new JavaMailSenderImpl();
sender.setHost("mail.coherentbank.com");
sender.send(message);
}
public void readExternal(PofReader pofReader) throws IOException {
account = (Account) pofReader.readObject(0);
}
public void writeExternal(PofWriter pofWriter) throws IOException {
pofWriter.writeObject(0, account);
}
}

As you can see, the run method simply uses the information from the account, such as customer's e-mail address and current balance, to create and send an e-mail message. In this example we used utility classes provided by Spring Framework in order to make the code more readable, but you could use plain JavaMail API to achieve the same thing.

You should also note that our task implements the PortableObject interface. This is necessary because the task will likely be executed by a remote cluster member, so Coherence needs to be able to serialize it in order to send it across the network.

Now that we have a Work implementation defined, we need to schedule it for execution. In order to do that, we need to obtain an instance of a WorkManager and pass our task to its schedule method:

WorkManager wm = new WorkManager("NotificationManager", 3);
wm.schedule(new LowBalanceAlertTask(acct1));
wm.schedule(new LowBalanceAlertTask(acct2));

The first argument passed to the WorkManager constructor is the name we want to use for this work manager.

Coherence WorkManager is implemented on top of Invocation Service, so the specified name will also be used as the name of the underlying InvocationService. One important caveat you should be aware of when creating WorkManager instances is that the WorkManager constructor can only be called once per Coherence node for any given work manager name. Any attempt to create an instance of a WorkManager that already exists will throw an exception. For that reason, you should store the reference to a created WorkManager in a static final variable, which will ensure that the constructor is called only once when the client class is initialized. If that's not possible, or you need to access the same WorkManager from multiple client classes, you can obtain it indirectly from the InvocationService it runs on top of:

Service service = CacheFactory.getService(managerName);
WorkManager wm = (WorkManager) service.getUserContext();

The second constructor argument is the number of worker threads the work manager should use to execute tasks. If you specify zero for this argument, WorkManager will be created in the "client" mode. This implies that it will only be able to schedule the work to be executed by the remote cluster members that belong to the same InvocationService. If you specify a negative integer, the default number of threads as defined in the cache configuration file will be used. This allows you to configure the thread pool size externally and is probably the best way to initialize WorkManager in your code.

When deciding on the number of threads to use, you should keep in mind that all work managers across the cluster that share a common name essentially form a cluster-wide thread pool to execute tasks on. That means that if we have a cluster of 10 nodes, each of which initializes our NotificationManager to use three threads, we will be able to execute 30 tasks simultaneously.

Once WorkManager is created, you can schedule tasks to execute by calling the schedule method, as we did in the example earlier. In addition to the method used in the example, there is an overload of the schedule method that allows you to pass a WorkListener to be notified as the status of the task changes.

The schedule method returns a WorkItem instance that can be used to determine the result of Work execution and to wait for completion of one or all tasks. In the example earlier, we didn't really care if the tasks have completed and what the result of each one was, so we simply ignored the return values. In cases when you do care about the results, you will need to do things a bit differently.

Coherence Work Manager implementation allows you to process the results of distributed Work execution using the same simple API you would use in a single-server application. All you need to do is to capture each WorkItem returned by the schedule method and use them to wait for the execution to finish and to get the result for each task.

For example, if we wanted to modify the example earlier to wait for all the notifications to be sent, we could implement it like this:

Set items = new HashSet();
WorkManager wm = new WorkManager("NotificationManager", 3);
items.add(wm.schedule(new LowBalanceAlertTask(acct1)));
items.add(wm.schedule(new LowBalanceAlertTask(acct2)));
...
wm.waitForAll(items, WorkManager.INDEFINITE);

As you can see, all we had to do was to add each WorkItem returned by the schedule method to the items collection and pass that collection to the WorkManager.waitForAll method. The second argument to waitForAll is the time in milliseconds to wait for all the tasks to complete. In this example, we have used one of the constants defined in the WorkManager interface to wait indefinitely until all the tasks are completed.

If you wanted to wait until any one of the scheduled tasks is completed, you could use the waitForAny method instead. An example of when this would be useful is when you have multiple ways to accomplish some task. For example, you might want to implement an alert notification that can be sent either via e-mail or SMS, but you only need to ensure that one of them is complete before proceeding. To achieve that, you would schedule both e-mail and SMS notification tasks and use waitForAny to block until at least one of them completes.

Once the task is marked as completed, you need to check what the result of its execution was by calling the getResult method on the related WorkItem. This method will either return the final Work instance, which might have been modified during execution, or will throw an exception if the task's run method threw an exception.