Salmon Run: jms

Showing posts with label jms. Show all posts

Saturday, August 14, 2010

A Recipe for Parallelization with Actors and JMS

Some time back, I spent some time researching various Actor Frameworks (you can find the posts here, here, here and here). While it was interesting, it was mostly academic for me, since they all implemented the Actor pattern in concurrent environments, ie, using threads in a single JVM, and I don't have access to large multi-core machines that these frameworks seemed to be aimed at.

Almost a year and a half later, a comment on one of these posts caused me to reread what I had written, and I realized that it may be possible to implement a parallel solution (distributing a large job across multiple low/medium powered machines) using the Actor pattern and JMS. As proof of concept, I decided to implement my little example of the 3-task pipeline (Download, Index and Write) that I had used for the other Actor examples.

Of course, since there does not exist a framework (none that I know of anyway) that does this, I built my own. Its called Kabuki, a Japanese theaterical art-form. Code wise, it is basically 1 abstract class with 2 inner classes that takes care of the JMS aspects, and exposes hooks (in the form of abstract methods) that an Actor subclass must implement.

For those of you who are not too familiar with what Actors (as defined in the Actor model) are, here is a nice definition from Concurrency in Erlang & Scala: The Actor Model by Ruben Vermeersch.

In the actor model, each object is an actor. This is an entity that has a mailbox and a behaviour. Messages can be exchanged between actors, which will be buffered in the mailbox. Upon receiving a message, the behaviour of the actor is executed, upon which the actor can: send a number of messages to other actors, create a number of actors and assume new behaviour for the next message to be received.

Like the actor defined above, Kabuki Actor subclasses define their own behavior (ie, the transformation on the incoming message to produce an outgoing message), and specify the location of a JMS queue that is to serve as its Inbox. Unlike it, however, the Actor does not decide (at runtime) where to send the outgoing message. Kabuki Actors have to specify at startup the location of a JMS queue to serve as its Outbox (if needed). The JMS queues are exposed by a JMS broker, Apache ActiveMQ in my case.

Kabuki Actor API

The Kabuki Actor superclass has 4 abstract methods which a subclass needs to implement. Sometimes these can be empty implementations. The table below lists these methods and a brief description.

Method-Name	Description
type()::Actor.Type	Kabuki Actors can be WRITE_ONLY, READ_WRITE or READ_ONLY. Determines whether the Actor only writes to the Outbox, reads from Inbox and writes to Outbox, and reads from Inbox respectively. All Actors must have an Inbox reference, but READ_ONLY Actors don't need an Outbox reference.
init()::void	Any application specific initialization, such as grabbing handles to external resources such as databases, files, etc. This is called once during the lifetime of the Kabuki Actor, when it is starting up.
perform(I input)::O	Implements the Kabuki Actor's behavior. Consumes an object of type I and transforms it into an object of type O. Called once for each message placed in the Actor's Inbox.
destroy()::void	Any application specific cleanup, such as flushing or releasing handles to external resources. Called once during the lifetime of the Kabuki Actor, when it is shutting down.

The pipeline of Actors for my example looks something like this:

In the example above, the Download Actor is the initiator (a WRITE_ONLY Actor). Although it has an Inbox, the only time it is ever used is when it calls the shutdown() method on itself. The Download Actor (a READ_WRITE Actor) writes to the Inbox of the Index Actor, and the Index Actor writes to the Inbox of the Write Actor (a READ_ONLY Actor). The Write Actor produces the output of the parallel job, and doesn't have an Outbox.

In our example, the Index Actor is the bottleneck, so we can alleviate the problem by starting up multiple instances of it. Since both Index Actors share the Inbox (a Queue exposed by the JMS broker), messages written by the Download Actor can be consumed by one of the Index Actors in the tier.

To illustrate how simple it is to actually implement Kabuki Actors, I show below the code for the three Actors in the chain. All they do is modify a String generated by the DownloadActor as it passes through the chain, but obviously you can make them do whatever you want them to.

DownloadActor.java

The DownloadActor generates the data. Obviously, you can dispense with this Actor and use some kind of Master process to pump data into the Actor pipeline. However, in this case, the JMS abstraction leaks a bit, since you (the programmer) is now forced to deal with the JMS message format. The DownloadActor demonstrates how to use a Kabuki Actor to generate the data - as you can see, the perform() method does not use its input parameter - it simply generates the data in a for loop. In a real world scenario, it would perhaps connect to an external data source and run some kind of filter on it periodically and pump the filtered data through.

The Download Actor (or any WRITE_ONLY Actor) is the only one in the pipeline that is allowed to call send(). Other Actors will do this function internally (in the superclass).

In lots of situations, Actors will run forever, but in our case, we want to terminate all the Actors in the pipeline once it pumps the data through. Once again, in something unique to WRITE_ONLY Actors, it calls shutdown(), which pumps a command through the pipeline that causes the other Actors to terminate as well.

// Source: src/main/java/com/mycompany/kabuki/actors/DownloadActor.java
package com.mycompany.kabuki.actors;

import com.mycompany.kabuki.core.Actor;

public class DownloadActor extends Actor<String,String> {

  @Override public Type type() {
    return Type.WRITE_ONLY;
  }

  @Override public void init() throws Exception {
    perform(null);
  }

  @Override public void destroy() throws Exception { /* NOOP */ }

  @Override public String perform(String input) throws Exception {
    for (int i = 0; i < 10; i++) {
      input = "Download Document-#:" + i;
      logger.info(input);
      send(input);
    }
    shutdown();
    return null;
  }
}

IndexActor.java

The IndexActor is a READ_WRITE Actor. It reads input messages off its Inbox, processes it using its perform() method, then places output messages on the Inbox of the next Actor in the pipeline.

// Source: src/main/java/com/mycompany/kabuki/actors/IndexActor.java
package com.mycompany.kabuki.actors;

import org.apache.commons.lang.StringUtils;

import com.mycompany.kabuki.core.Actor;

public class IndexActor extends Actor<String,String> {

  @Override public Type type() {
    return Type.READ_WRITE;
  }

  @Override public void init() throws Exception { /* NOOP */ }

  @Override public void destroy() throws Exception { /* NOOP */ }

  @Override public String perform(String input) {
    String output = StringUtils.replace(input, "Download", "Index");
    logger.info(output);
    return output;
  }
}

WriteActor.java

The WriteActor is the final step in my pipeline. Since its function is to write out the text file containing the value of the converted String, it needs to open and close the handle to the file - this is done in the init() and destroy() methods. The other Actors in my chain did not need to do this, so their init() and destroy() methods are empty.

The WriteActor also does not forward the converted data to the next Actor in the pipeline, because there is no other Actor.

// Source: src/main/java/com/mycompany/kabuki/actors/WriteActor.java
package com.mycompany.kabuki.actors;

import java.io.FileWriter;
import java.io.PrintWriter;

import org.apache.commons.lang.StringUtils;

import com.mycompany.kabuki.core.Actor;

public class WriteActor extends Actor<String,String> {

  PrintWriter printWriter;
  
  @Override public Type type() {
    return Type.READ_ONLY;
  }

  @Override public void init() throws Exception {
    printWriter = new PrintWriter(new FileWriter("/tmp/demo.txt"), true);
  }

  @Override public void destroy() throws Exception {
    if (printWriter != null) {
      printWriter.flush();
      printWriter.close();
    }
  }

  @Override public String perform(String input) {
    String output = StringUtils.replace(input, "Index", "Write");
    logger.info(output);
    printWriter.println(output);
    return null;
  }
}

Command line syntax

The pipeline can be set up as a shell script. Currently my script runs all the Actors on the same box, but we can use passwordless SSH to start up Actors on different machines. Here is my script:

#!/bin/bash
PROJECT_HOME=/Users/sujit/Projects/kabuki
CLASSPATH=\
  $PROJECT_HOME/lib/commons-lang-2.5.jar:\
  $PROJECT_HOME/lib/commons-cli-1.0.jar:\
  $PROJECT_HOME/lib/activemq-all-5.3.2.jar:\
  $PROJECT_HOME/lib/log4j-1.2.14.jar:\
  $PROJECT_HOME/target/kabuki.jar
BROKER_URL=tcp://localhost:61616
LOG4J_CONFIG=file:$PROJECT_HOME/src/main/resources/log4j.properties
# single instance of DownloadActor
java -cp $CLASSPATH -Dlog4j.configuration=$LOG4J_CONFIG \
  com.mycompany.kabuki.core.Actor \
  -a com.mycompany.kabuki.actors.DownloadActor \
  -u $BROKER_URL -i download -o index &
# multiple parallel instances of IndexActor
java -cp $CLASSPATH -Dlog4j.configuration=$LOG4J_CONFIG \
  com.mycompany.kabuki.core.Actor \
  -a com.mycompany.kabuki.actors.IndexActor \
  -u $BROKER_URL -i index -o write &
java -cp $CLASSPATH -Dlog4j.configuration=$LOG4J_CONFIG \
  com.mycompany.kabuki.core.Actor \
  -a com.mycompany.kabuki.actors.IndexActor \
  -u $BROKER_URL -i index -o write &
# single instance of WriteActor
java -cp $CLASSPATH -Dlog4j.configuration=$LOG4J_CONFIG \
  com.mycompany.kabuki.core.Actor \
  -a com.mycompany.kabuki.actors.WriteActor \
  -u $BROKER_URL -i write &
wait

As you can see, you assign the inbox and outbox to different Actors by using the -i and -o parameters. The script assumes that the JMS broker has been set up. For the JMS Broker, I used a stock Apache ActiveMQ - just downloaded it and started it with the default configuration.

Kabuki Internals

As you know, the Kabuki "framework" consists of a single Actor class that all Kabuki Actors must extend. Here is the code for the Actor superclass:

// Source: src/main/java/com/mycompany/kabuki/core/Actor.java
package com.mycompany.kabuki.core;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

/**
 * Superclass that all Actors implementations must extend.
 * @author Sujit Pal (spal@healthline.com)
 * @version $Revision$
 */
public abstract class Actor<I,O> {

  public static enum Type {
    READ_ONLY, WRITE_ONLY, READ_WRITE
  };
  
  public static final String SHUTDOWN_COMMAND = "SHUTDOWN";

  private static final String PAYLOAD_KEY = "payload";
  private static final String COMMAND_KEY = "command";
  private static final long SHUTDOWN_DELAY = 5000; // 5s
  
  protected final Logger logger = Logger.getLogger(getClass());
  
  private String brokerUrl;
  private String inbox;
  private String outbox;
  
  private Connection connection;
  // payload is p2p - since we want only one actor at a tier to pick
  // up the incoming payload
  private Session payloadSession;
  private Queue payloadInbox;
  private Queue payloadOutbox;
  private MessageConsumer payloadReceiver;
  private MessageProducer payloadSender;
  // command is pub-sub, since we want all actors in a tier to pick
  // up the incoming command
  private Session commandSession;
  private Topic commandTopic;
  private MessageConsumer commandSubscriber;
  private MessageProducer commandPublisher;
  
  public abstract Type type();
  public abstract void init() throws Exception;
  public abstract O perform(I input) throws Exception;
  public abstract void destroy() throws Exception;

  public void setBrokerUrl(String brokerUrl) {
    this.brokerUrl = brokerUrl;
  }
  
  public void setInbox(String inbox) {
    this.inbox = inbox;
  }
  
  public void setOutbox(String outbox) {
    this.outbox = outbox;
  }
  
  protected final void start() throws Exception {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);
    connection = factory.createConnection();
    commandSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    payloadSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    // set up payload queue destinations and friends
    payloadInbox = payloadSession.createQueue(
      StringUtils.join(new String[] {PAYLOAD_KEY, inbox}, "."));
    payloadReceiver = payloadSession.createConsumer(payloadInbox);
    payloadReceiver.setMessageListener(new PayloadListener());
    if (type() == Type.WRITE_ONLY || type() == Type.READ_WRITE) {
      payloadOutbox = payloadSession.createQueue(
        StringUtils.join(new String[] {PAYLOAD_KEY, outbox}, "."));
      payloadSender = payloadSession.createProducer(payloadOutbox);
    }
    // set up command topic destination and friends
    commandTopic = commandSession.createTopic(
      StringUtils.join(new String[] {
      COMMAND_KEY, getClass().getSimpleName()}, ".")); 
    commandSubscriber = commandSession.createConsumer(commandTopic);
    commandSubscriber.setMessageListener(new CommandListener());
    commandPublisher = commandSession.createProducer(commandTopic);
    // start your engine
    connection.start();
    logger.info("Actor Started: " + getClass().getName());
    // call the init() hook
    init();
  }

  protected void shutdown() throws Exception {
    if (payloadOutbox != null) {
      MapMessage shutdownMessage = payloadSession.createMapMessage();
      shutdownMessage.setString(COMMAND_KEY, SHUTDOWN_COMMAND);
      payloadSender.send(shutdownMessage);
    }
    if (type() == Type.WRITE_ONLY) {
      logger.info("Waiting " + SHUTDOWN_DELAY + "ms...");
      try { Thread.sleep(SHUTDOWN_DELAY); }
      catch (InterruptedException e) { /* NOOP */ }
      stop();
    }
  }

  protected final void send(O output) throws Exception {
    if (type() != Type.WRITE_ONLY) {
      throw new IllegalAccessError(
        "send() can only be called from WRITE_ONLY Actors");
    }
    if (payloadSender != null) {
      MapMessage payload = payloadSession.createMapMessage();
      payload.setObject(PAYLOAD_KEY, output);
      payloadSender.send(payload);
    }
  }

  private final void stop() throws Exception {
    closeQuietly(payloadReceiver);
    closeQuietly(commandSubscriber);
    closeQuietly(payloadSender);
    closeQuietly(commandPublisher);
    commandSession.close();
    payloadSession.close();
    connection.close();
    // call the destroy hook
    destroy();
    logger.info("Actor stopped: " + getClass().getName());
  }

  private void closeQuietly(MessageProducer producer) {
    if (producer != null) {
      try { producer.close(); }
      catch (JMSException e) { /* NOOP */ }
    }
  }
  
  private void closeQuietly(MessageConsumer consumer) {
    if (consumer != null) {
      try { consumer.close(); }
      catch (JMSException e) { /* NOOP */ }
    }
  }
  
  private final class PayloadListener implements MessageListener {
    @SuppressWarnings("unchecked")
    @Override public void onMessage(Message message) {
      try {
        MapMessage payload = (MapMessage) message;
        // check to see if this is a shutdown message. If so, send to
        // the command topic for this actor 
        String command = (String) payload.getObject(COMMAND_KEY);
        if (SHUTDOWN_COMMAND.equals(command)) {
          // send the shutdown command to the command topic
          TextMessage shutdownMessage = commandSession.createTextMessage();
          shutdownMessage.setText(SHUTDOWN_COMMAND);
          commandPublisher.send(shutdownMessage);
          // pass it on to the next tier
          shutdown();
        } else {
          I input = (I) payload.getObject(PAYLOAD_KEY);
          O output = null;
          if (input != null) {
            try { 
              output = perform(input);
            } catch (Exception e) {
              logger.error(e);
            }
          }
          if (output != null && payloadSender != null) {
            MapMessage outputPayload = payloadSession.createMapMessage();
            outputPayload.setObject(PAYLOAD_KEY, output);
            payloadSender.send(outputPayload);
          }
        }
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  };

  private final class CommandListener implements MessageListener {
    @Override public void onMessage(Message message) {
      try {
        TextMessage command = (TextMessage) message;
        if (SHUTDOWN_COMMAND.equals(command.getText())) {
          stop();
        }
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  };

  // these methods are called from the shell script to start up an instance
  // of an Actor.
  
  @SuppressWarnings("unchecked")
  public static void main(String[] args) throws Exception {
    CommandLineParser clp = new BasicParser();
    Options options = new Options();
    options.addOption("h", "help", false, "Print this message");
    options.addOption("a", "actor", true, "Full class name for Actor");
    options.addOption("i", "input", true, "Inbox");
    options.addOption("o", "output", true, "Outbox (optional for READ_ONLY)");
    options.addOption("u", "brokerUrl", true, "ActiveMQ Broker URL");
    CommandLine cl = clp.parse(options, args);
    if (cl.hasOption("h")) {
      printUsage(null, options);
      System.exit(0);
    }
    String actorClassName = null;
    if (cl.hasOption("a")) {
      actorClassName = cl.getOptionValue("a");
    }
    String brokerUrl = null;
    if (cl.hasOption("u")) {
      brokerUrl = cl.getOptionValue("u");
    }
    String inputAlias = null;
    if (cl.hasOption("i")) {
      inputAlias = cl.getOptionValue("i");
    }
    String outputAlias = null;
    if (cl.hasOption("o")) {
      outputAlias = cl.getOptionValue("o");
    }
    // validation: broker url must be defined
    if (StringUtils.isEmpty(brokerUrl)) {
      printUsage("Broker URL must be defined", options);
    }
    // validation: actor class name must be defined
    if (StringUtils.isEmpty(brokerUrl)) {
      printUsage("Actor class must be defined", options);
    }
    Actor actor = 
      (Actor) Class.forName(actorClassName).newInstance();
    // validation: 
    // all actors must have a payload inbox
    // only read-only actors may or may not have a payload outbox
    if (StringUtils.isEmpty(inputAlias)) {
      printUsage("No Inbox specified for actor:" + actorClassName, options);
    }
    if (StringUtils.isEmpty(outputAlias)) {
      if (actor.type() != Type.READ_ONLY) {
        printUsage("No Outbox specified for actor:" + actorClassName, options);
      }
    }
    actor.setBrokerUrl(brokerUrl);
    actor.setInbox(inputAlias);
    actor.setOutbox(outputAlias);
    actor.start();
  }

  private static void printUsage(String message, Options options) {
    if (StringUtils.isNotEmpty(message)) {
      System.out.println("ERROR: " + message);
    }
    HelpFormatter formatter = new HelpFormatter();
    formatter.defaultWidth = 80;
    formatter.printHelp("java " + Actor.class.getName() + 
      " [-h|-a class -u url -i input [-o output]]", options);
  }
}

Internally, each Actor sets up two sessions, one for publish-subscribe messaging and another for point-to-point messaging. Both sessions have asynchronouse Listeners (MessageListeners) which listen for messages on the incoming Queue or Topic until the sessions are closed.

The point-to-point session is used by the Actor to read incoming messages off the Inbox Queue and write the output of perform() onto its Outbox Queue. Point-to-point ensures that a mesage is processed by an Actor only once.

The publish-subscribe session is used for sending the SHUTDOWN command - if there are multiple instances of an Actor at a particular tier in the pipeline, then we want a single SHUTDOWN command to percolate to all the Actors in the tier. This is achieved by setting up a single command Topic per tier (using the simple class name of the actor as part of the Topic name) and having all the Actors in this tier subscribe to the Topic.

Ordering (ie, the SHUTDOWN should be processed after all payload is consumed) is achieved by sending the SHUTDOWN command in the same format as the payload (ie a MapMessage). The payload listener checks to see if the message is a SHUTDOWN message, and if so, writes out a message to the command Topic, which is then consumed by the command listener on every Actor in that tier, causing them to terminate.

Conclusion

I haven't tried this out with a real world scenario yet, but my demo seem to work fine, which is encouraging. While I have used JMS before and also played with various Actor framework flavors, I am by no means an authority on either subject. If you are, or if you notice flaws in my design, would appreciate your letting me know.

Saturday, April 12, 2008

OSWorkflow meets JMS

The title of this post is probably unfair, since the authors of OSWorkflow provide a built-in JMSMessage action to do Fire-and-Forget style calls to a JMS Queue or Topic. However, what I describe here takes that integration one step further, by marrying event-based OSWorkflow processing with Asynchronous with Callback style JMS calls using Apache ActiveMQ. My application needs to fire long running batch job which are dependent on each other. Rather than have a human operator fire them off in the right sequence, the idea is to build a workflow that captures these dependencies, then submit them asynchronously to a JMS queue. As each job complete, the JMS listener which executes these jobs on the other end sends callbacks to the workflow, which allow it to fire the next job in the dependency graph.

The workflow is fairly complex, it contains two splits and two corresponding joins, as shown below. See my previous post for a more thorough discussion of the workflow itself. I have reproduced the graph below for completeness below:

There are three main components in here. First the WorkflowRunner, which loads the OSWorkflow configuration for the workflow into memory. It is called once from an external client (probably a web user) to kick off the workflow. Each workflow step that is responsible for executing a batch job is tied to a custom JmsAction which writes a message to the request.topic Topic. At the far end of the Topics is a JmsServer component, which basically delegates off to Java processes representing each individual batch job. Once a job is complete, it writes a message to the response.topic Topic. This message is picked up by the correct JmsAction, which then publishes an event into the Spring ApplicationContext. The WorkflowRunner then picks this event up, and issues calls to process() recursively until the workflow is complete. The diagram below illustrates this flow.

I describe below each of the individual components. First, the OSWorkflow workflow definition file. Each of the steps are mapped to a JmsAction instance. This is the only difference from the previous week, when everything was mapped to a MockAction instance.

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE workflow PUBLIC 
  "-//OpenSymphony Group//DTD OSWorkflow 2.8//EN"
  "http://www.opensymphony.com/osworkflow/workflow_2_8.dtd">
<workflow>
  <initial-actions>
    <action id="0" name="start">
      <pre-functions>
        <function type="class">
          <arg name="class.name">com.opensymphony.workflow.util.Caller</arg>
        </function>
        <function type="spring">
          <arg name="bean.name">jmsAction</arg>
          <arg name="action.name">t0</arg>
        </function>
      </pre-functions>
      <results>
        <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
          step="1"/>
      </results>
    </action>
  </initial-actions>
  <steps>
    <step id="1" name="p1">
      <actions>
        <action id="1" name="t1">
          <results>
            <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
              step="26"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t1</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="26" name="s26">
      <actions>
        <action id="26" name="split26">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" split="26"/>
          </results>
        </action>
      </actions>
    </step>
    <step id="2" name="p2">
      <actions>
        <action id="2" name="t2">
          <results>
            <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
              step="34"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t2</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="34" name="s34">
      <actions>
        <action id="34" name="split34">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" split="34"/>
          </results>
        </action>
      </actions>
    </step>
    <step id="3" name="p3">
      <actions>
        <action id="3" name="t3">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" join="43"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t3</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="4" name="p4">
      <actions>
        <action id="4" name="t4">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" join="43"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t4</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="5" name="p5">
      <actions>
        <action id="5" name="t5">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" join="75"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t5</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="6" name="p6">
      <actions>
        <action id="6" name="t6">
          <results>
            <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
              step="7"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t6</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="7" name="p7">
      <actions>
        <action id="7" name="t7">
          <results>
            <unconditional-result old-status="Finished" owner="${caller}" join="75"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t7</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="8" name="p8">
      <actions>
        <action id="8" name="t8">
          <results>
            <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
              step="9"/>
          </results>
          <post-functions>
            <function type="spring">
              <arg name="bean.name">jmsAction</arg>
              <arg name="action.name">t8</arg>
            </function>
          </post-functions>
        </action>
      </actions>
    </step>
    <step id="9" name="stop">
      <actions>
        <action id="9" name="t9" finish="true">
          <results>
            <unconditional-result old-status="Finished" status="Complete" 
              owner="${caller}"/>
          </results>
        </action>
      </actions>
    </step>
  </steps>
  <splits>
    <split id="26">
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="2"/>
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="6"/>
    </split>
    <split id="34">
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="3"/>
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="4"/>
    </split>
  </splits>
  <joins>
    <join id="43">
      <conditions type="AND">
        <condition type="beanshell">
          <arg name="script"><![CDATA[
            "Finished".equals(jn.getStep(3).getStatus()) &&
            "Finished".equals(jn.getStep(4).getStatus())
          ]]>
          </arg>
        </condition>
      </conditions>
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="5"/>
    </join>
    <join id="75">
      <conditions type="AND">
        <condition type="beanshell">
          <arg name="script"><![CDATA[
          "Finished".equals(jn.getStep(5).getStatus()) &&
          "Finished".equals(jn.getStep(7).getStatus())
          ]]>
          </arg>
        </condition>
      </conditions>
      <unconditional-result old-status="Finished" status="Queued" owner="${caller}" 
        step="8"/>
    </join>
  </joins>
</workflow>

Next up is the Spring applicationContext.xml file which ties this all together. It also contains some definitions for the JmsAction and JmsServer beans. This is shown below:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:util="http://www.springframework.org/schema/util"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
       http://www.springframework.org/schema/util 
       http://www.springframework.org/schema/util/spring-util-2.0.xsd">

  <bean class="org.springframework.beans.factory.annotation.RequiredAnnotationBeanPostProcessor"/>

  <!-- OSWorkflow -->
  <bean id="workflowStore" 
    class="com.opensymphony.workflow.spi.memory.MemoryWorkflowStore"/>
  
  <bean id="workflowFactory" 
      class="com.opensymphony.workflow.spi.hibernate.SpringWorkflowFactory" 
      init-method="init">
    <property name="resource" value="workflow-defs.xml"/>
    <property name="reload" value="true"/>    
  </bean>

  <bean id="workflowConfiguration" 
      class="com.opensymphony.workflow.config.SpringConfiguration">
    <property name="factory" ref="workflowFactory"/>
    <property name="store" ref="workflowStore"/>
  </bean>
  
  <bean id="workflowTypeResolver" class="com.opensymphony.workflow.util.SpringTypeResolver">
    <property name="functions">
      <map>
        <entry key="jmsAction"><ref bean="jmsAction"/></entry>
      </map>
    </property>
  </bean>

  <bean id="workflow" class="com.opensymphony.workflow.basic.BasicWorkflow" 
      scope="prototype">
    <constructor-arg><value>testuser</value></constructor-arg>
    <property name="configuration" ref="workflowConfiguration"/>
    <property name="resolver" ref="workflowTypeResolver"/>
  </bean>
    
  <bean id="workflowRunner" class="com.mycompany.myapp.workflow.WorkflowRunner" 
      scope="prototype">
    <property name="workflow" ref="workflow"/>
  </bean>
  
  <!-- JMS -->
  <bean id="jmsConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" 
      destroy-method="stop">
    <property name="connectionFactory">
      <bean class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL">
          <value>tcp://localhost:61616</value>
        </property>
      </bean>
    </property>
  </bean>

  <bean id="jmsAction" class="com.mycompany.myapp.workflow.JmsAction" init-method="init" 
      destroy-method="destroy" scope="prototype">
    <property name="jmsConnectionFactory" ref="jmsConnectionFactory"/>
  </bean>

  <bean id="jmsServer" class="com.mycompany.myapp.workflow.JmsServer">
    <property name="jmsConnectionFactory" ref="jmsConnectionFactory"/>
  </bean>
    
</beans>

As mentioned above, on the OSWorkflow side, we have a WorkflowRunner component. It implements ApplicationListener so it can respond to callbacks from the JmsAction components. Moving through the workflow takes place in the process() method. The onApplicationEvent() recursively calls the process() method based on events received from the JmsAction instances.

// WorkflowRunner.java
package com.mycompany.myapp.workflow;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;

import com.opensymphony.workflow.Workflow;
import com.opensymphony.workflow.WorkflowException;
import com.opensymphony.workflow.spi.Step;

/**
 * This class is called once by the client code, then the events that are
 * sent back from the Actions in the Workflow will move the workflow forward
 * until it ends.
 */
public class WorkflowRunner implements ApplicationListener {

  private final Log log = LogFactory.getLog(getClass());
  
  private String workflowName;
  private Workflow workflow;
  private Map<String,Object> inputs;
  
  private Set<Integer> alreadyExecuted = new HashSet<Integer>();
  
  @Required
  public void setWorkflow(Workflow workflow) {
    this.workflow = workflow;
  }
  
  public void setWorkflowName(String workflowName) {
    this.workflowName = workflowName;
  }
  
  public void setInputs(Map<String,Object> inputs) {
    this.inputs = inputs;
  }
  
  public long init(int initialActionId) throws Exception {
    long workflowId = workflow.initialize(workflowName, initialActionId, inputs);
    return workflowId;
  }
  
  @SuppressWarnings({"unused","unchecked"})
  public void process(long workflowId) {
    List<Step> currentSteps = workflow.getCurrentSteps(workflowId);
    for (Step currentStep : currentSteps) {
      int[] availableActions = workflow.getAvailableActions(workflowId, inputs);
      for (int availableAction : availableActions) {
        if (alreadyExecuted.contains(availableAction)) {
          continue;
        }
        try {
          alreadyExecuted.add(availableAction);
          System.out.println("Sending action.id=" + availableAction + " to JmsServer");
          workflow.doAction(workflowId, availableAction, inputs);
        } catch (WorkflowException e) {
          log.error("Exception in (workflow,action.id)=(" + workflowName + "," + 
            availableAction + "). Workflow stopped", e);
        }
      }
    }
  }
  
  public void onApplicationEvent(ApplicationEvent event) {
    if (event.getSource() instanceof Long) {
      process((Long) event.getSource());
    }
  }
}

Not too many surprises so far for people who have read my previous blog post. The JmsAction is our bridge between OSWorkflow and JMS, so it has to implement multiple interfaces. The FunctionProvider interface is so it can be injected as a function into the workflow XML configuration. ApplicationContextAware is so it can publish events back into Spring's context where it can be picked up by WorkflowRunner, and MessageListener is so it can listen on JMS events. The code is shown below and is fairly self-explanatory, comments are inlined where I felt more explanation may be helpful.

// JmsAction.java
package com.mycompany.myapp.workflow;

import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;

import com.opensymphony.module.propertyset.PropertySet;
import com.opensymphony.workflow.FunctionProvider;
import com.opensymphony.workflow.WorkflowException;
import com.opensymphony.workflow.spi.WorkflowEntry;

/**
 * Publishes a job request to a JMS queue. Listens for a callback from the
 * JMS consumer module and passes the event back to the WorkflowRunner event
 * listener to move the workflow forward.
 */
public class JmsAction implements FunctionProvider, ApplicationContextAware, MessageListener {

  private final Log log = LogFactory.getLog(getClass());
  
  private ConnectionFactory jmsConnectionFactory;
  private String actionName;
  private ApplicationContext applicationContext;
  
  private Connection connection;
  private Session session;
  private MessageProducer messageProducer;
  private MessageConsumer messageConsumer;

  @Required
  public void setJmsConnectionFactory(ConnectionFactory jmsConnectionFactory) {
    this.jmsConnectionFactory = jmsConnectionFactory; 
  }

  public void setActionName(String actionName) {
    this.actionName = actionName;
  }

  /**
   * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
   */
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    this.applicationContext = applicationContext;
  }

  /**
   * @see com.opensymphony.workflow.FunctionProvider#execute(java.util.Map, java.util.Map, com.opensymphony.module.propertyset.PropertySet)
   */
  @SuppressWarnings("unchecked")
  public void execute(Map transientVars, Map args, PropertySet ps) throws WorkflowException {
    setActionName((String) args.get("action.name"));
    WorkflowEntry workflowEntry = (WorkflowEntry) transientVars.get("entry");
    final long workflowId = workflowEntry.getId();
    try {
      MapMessage message = session.createMapMessage();
      message.setString("action.name", actionName);
      message.setString("workflow.id", String.valueOf(workflowId));
      message.setString("topic.name", "request.topic");
      messageProducer.send(message);
    } catch (JMSException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * Listens for a text message back from the JmsSubscriber module about job
   * completion. The text is formatted, contains Success or Failure, followed
   * by workflowId as a comma-separated list.
   * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
   */
  @SuppressWarnings("serial")
  public void onMessage(Message message) {
    if (MapMessage.class.isInstance(message)) {
      MapMessage mapMessage = MapMessage.class.cast(message);
      try {
        String actionName = mapMessage.getString("action.name");
        if (actionName.equals(this.actionName)) {
          // only respond to events meant for this Action
          String status = mapMessage.getString("status");
          String workflowId = mapMessage.getString("workflow.id");
          if (status.equals("Success")) {
            applicationContext.publishEvent(new ApplicationEvent(new Long(workflowId)) {});
          } else {
            log.error("Action " + actionName + " failed, see server error log for details");
          }
        }
      } catch (JMSException e) {
        throw new RuntimeException(e);
      }
    }
  }
  
  protected void init() throws Exception {
    connection = jmsConnectionFactory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic requestTopic = session.createTopic("request.topic");
    Topic responseTopic = session.createTopic("response.topic");
    messageProducer = session.createProducer(requestTopic);
    messageConsumer = session.createConsumer(responseTopic);
    messageConsumer.setMessageListener(this);
    connection.start();
  }
  
  protected void destroy() throws Exception {
    session.close();
    connection.close();
  }
}

I started off using Spring's JmsTemplate because I wanted to learn how to use it, but gave up when I could not find a clean way of registering a JmsAction as a listener. If any of you have used JmsTemplate for beans which are both publishers and subscribers, please let me know. Example code (or links to example code) would be greatly appreciated.

One more thing to note is that unlike the WorkflowRunner, which will have only a single instance per workflow, there will be many instances of JmsAction for a given workflow. We specify that its scope is prototype (built every time it is accessed from the Spring context), and that its lifecycle methods are init() and destroy(), all in the Spring configuration above.

Our final component is the JmsServer. All this does currently is to print that it is "executing" something to stdout. In a real application, it would start another Java batch process and wait for it to complete before sending the callback. The code is shown below:

// JmsServer.java
package com.mycompany.myapp.workflow;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;

/**
 * Listens for JMS requests, services them, and returns a callback
 * when the job is completed.
 */
public class JmsServer implements MessageListener {

  private ConnectionFactory jmsConnectionFactory;
  
  private Connection connection;
  private Session session;
  private MessageProducer messageProducer;
  private MessageConsumer messageConsumer;

  public void setJmsConnectionFactory(ConnectionFactory jmsConnectionFactory) {
    this.jmsConnectionFactory = jmsConnectionFactory;
  }
  
  /**
   * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
   */
  public void onMessage(Message message) {
    if (! MapMessage.class.isInstance(message)) {
      return;
    }
    try {
      MapMessage mapMessage = MapMessage.class.cast(message);
      final String actionName = mapMessage.getString("action.name");
      final String workflowId = mapMessage.getString("workflow.id");
      // this is where we will delegate to some kind of executor in a real app
      System.out.println("Executing job:" + actionName);
      // send the callback after the job is done
      MapMessage responseMessage = session.createMapMessage();
      responseMessage.setString("action.name", actionName);
      responseMessage.setString("workflow.id", workflowId);
      responseMessage.setString("topic.name", "response.topic");
      responseMessage.setString("status", "Success");
      System.out.println("Sending callback from server");
      messageProducer.send(responseMessage);
    } catch (JMSException e) {
      throw new RuntimeException(e);
    }
  }
  
  protected void init() throws Exception {
    System.out.println("Initializing server");
    connection = jmsConnectionFactory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic responseTopic = session.createTopic("request.topic");
    Topic requestTopic = session.createTopic("response.topic");
    messageProducer = session.createProducer(requestTopic);
    messageConsumer = session.createConsumer(responseTopic);
    messageConsumer.setMessageListener(this);
    System.out.println("Server started");
    connection.start();
  }
  
  protected void destroy() throws Exception {
    System.out.println("Shutting down server");
    session.close();
    connection.close();
    System.out.println("Done");
  }
  
  public void run() throws Exception {
    init();
    try {
      for (;;) {
        Thread.sleep(500L);
      }
    } finally {
      destroy();
    }
  }
}

To test this, we start off the following two unit tests which run forever until they are manually stopped by a Ctrl-C at the command line. The reason I had to do this instead of packaging the whole thing into a single JUnit test as I had done before, was because I noticed that the unit test completed before the workflow had a chance to complete. Since the whole thing is event driven after the first call, there is no way to keep the test running until the workflow is complete. In real-life, this is not a problem because these are likely to be long-lived server processes anyway.

This JUnit test starts the JmsServer end of the setup and loops forever, sleeping for 0.5s between loops. I run this in one console window.

// JmsServerTest.java
package com.mycompany.myapp.workflow;

import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class JmsServerTest {

  private JmsServer jmsServer;
  
  @Before
  public void setUp() throws Exception {
     ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
     jmsServer = (JmsServer) context.getBean("jmsServer");
  }
  
  @Test
  public void runJmsServer() throws Exception {
    jmsServer.run();
  }
}

This JUnit test starts the OSWorkflow end of the setup and after the first call to process(), also loops forever. I run this in a second console window.

// WorkflowRunnerWithJmsActionsTest.java
package com.mycompany.myapp.workflow;

import java.util.HashMap;
import java.util.Map;

import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class WorkflowRunnerWithJmsActionsTest {

  private WorkflowRunner runner;
  private Map<String,Object> inputs = new HashMap<String,Object>();
  private long workflowId;

  @Before
  public void setUp() throws Exception {
    ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml"); 
    runner = (WorkflowRunner) context.getBean("workflowRunner");
    runner.setWorkflowName("workflow-def-1");
    runner.setInputs(inputs);
    workflowId = runner.init(0);
  }
  
  @Test
  public void runWorkflow() throws Exception {
    runner.process(workflowId);
    for (;;) {
      Thread.sleep(500L);
    }
  }
}

Running these produces the following output on the client end, and a similar output on the server end. Looking at the output, it is clear that the setup works, ie the output is what you would expect given the dependencies designed into the workflow.

Sending action.id=1 to JmsServer
Sending action.id=26 to JmsServer
Sending action.id=2 to JmsServer
Sending action.id=6 to JmsServer
Sending action.id=34 to JmsServer
Sending action.id=7 to JmsServer
Sending action.id=3 to JmsServer
Sending action.id=4 to JmsServer
Sending action.id=5 to JmsServer
Sending action.id=8 to JmsServer
Sending action.id=9 to JmsServer

A nice tool that helped me debug my code is ActiveMQ's web-based administrator interface. Here you can see how many messages were received and sent on each of the topics. You can delete messages off the queue, which is very helpful after unsuccessful runs. It is available at 0.0.0.0:8161 of the machine where ApacheMQ is started.

So there you have it folks. A simple, asynchronous, event-driven way to manage dependencies among multiple batch jobs.

Salmon Run

Saturday, August 14, 2010

A Recipe for Parallelization with Actors and JMS

Kabuki Actor API

Kabuki Internals

Conclusion

Saturday, April 12, 2008

OSWorkflow meets JMS

Posts

Labels

Blogs I Read

About me

My Nerd Rating

Visitor Map

Contact Me