commandexecutorthread.java

jBpm是一个灵活可扩展的工作流管理系统。作为jBpm运行时server输入的业务流程使用简单强大的语言表达并打包在流程档案中

/*
 * JBoss, Home of Professional Open Source
 * Copyright 2005, JBoss Inc., and individual contributors as indicated
 * by the @authors tag. See the copyright.txt in the distribution for a
 * full listing of individual contributors.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this software; if not, write to the Free
 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
 */
package org.jbpm.msg.command;

import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jbpm.JbpmConfiguration;
import org.jbpm.JbpmContext;
import org.jbpm.JbpmException;
import org.jbpm.command.Command;
import org.jbpm.configuration.ConfigurationException;
import org.jbpm.db.MessagingSession;
import org.jbpm.msg.Message;
import org.jbpm.msg.db.DbMessageService;
import org.jbpm.msg.db.StaticNotifier;

public class CommandExecutorThread extends Thread implements Serializable {

  private static final long serialVersionUID = 1L;

  public static final String DEFAULT_COMMAND_EXECUTOR_CONTEXT_NAME = JbpmContext.DEFAULT_JBPM_CONTEXT_NAME;
  public static final String DEFAULT_ERROR_DESTINATION = "ERROR";

  JbpmConfiguration jbpmConfiguration = null;
  String jbpmContextName = JbpmContext.DEFAULT_JBPM_CONTEXT_NAME;
  String destination = Command.DEFAULT_CMD_DESTINATION;
  String errorDestination = DEFAULT_ERROR_DESTINATION;

  public CommandExecutorThread(JbpmConfiguration jbpmConfiguration) {
    super("JbpmCommandExecutor");
    this.jbpmConfiguration = jbpmConfiguration;
    if (this.jbpmConfiguration==null) {
      throw new JbpmException("jbpmConfiguration is null");
    }
  }
  
  public void setDestination(String destination) {
    this.destination = destination;
  }
  public void setErrorDestination(String errorDestination) {
    this.errorDestination = errorDestination;
  }
  public void setJbpmConfiguration(JbpmConfiguration jbpmConfiguration) {
    this.jbpmConfiguration = jbpmConfiguration;
  }
  public void setJbpmContextName(String jbpmContextName) {
    this.jbpmContextName = jbpmContextName;
  }

  public void run() {
    // while not interrupted...
    try {
      while (true) {
        boolean checkForMoreMessages = true;
        try {
          checkForMoreMessages = executeCommand();

        } catch (MessageProcessingException e) {
          log.error(e);
          handleProcessingException(e);
        } catch (InterruptedException e) {
          throw e;
        } catch (Exception other) {
          // NOTE that Error's are not caught because that might halt the JVM and mask the original Error.
          log.error(other);
          Thread.sleep(5000);
        }
        
        if (! checkForMoreMessages) {
          log.debug("waiting for more messages");
          StaticNotifier.waitForNotification(Command.DEFAULT_CMD_DESTINATION);
        }
      }
    } catch (InterruptedException e) {
      log.info("jBPM command executor got interrupted");
    } finally {
      log.warn("jBPM command executor STOPPED");
    }
  }

  public boolean executeCommand() throws Exception {
    boolean checkForMoreMessages = false;
    Message message = null; 
    JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(jbpmContextName);
    try {
      // get a command from the queue
      DbMessageService dbMessageService = null;
      try {
        dbMessageService = (DbMessageService) jbpmContext.getServices().getMessageService();
      } catch (ClassCastException e) {
        throw new ConfigurationException("CommandExecutorThread only works with the DbMessageService implementation of the MessageService. please, configure jbpm.cfg.xml accordingly.");
      }
      if (dbMessageService==null) {
        throw new ConfigurationException("no messaging service available");
      }
      message = dbMessageService.receiveNoWait(destination);
      
      if (message!=null) {
        checkForMoreMessages = true;
        Command command = (Command) message;
        log.trace("executing command '"+command+"'");
        command.execute();
      }
      
    } catch (Exception e) {
      // rollback the transaction
      log.debug("command '"+message+"' threw exception. rolling back transaction", e);
      jbpmContext.setRollbackOnly();
      
      if (message!=null) {
        throw new MessageProcessingException(message, e);
      } else {
        throw e;
      }
    } finally {
      jbpmContext.close();
    }
    
    return checkForMoreMessages;
  }
  
  void handleProcessingException(MessageProcessingException e) {
    JbpmContext jbpmContext = jbpmConfiguration.createJbpmContext(jbpmContextName);
    try {
      // get the message session from the context
      DbMessageService dbMessageSessionImpl = (DbMessageService) jbpmContext.getServices().getMessageService();
      MessagingSession messageSession = dbMessageSessionImpl.getMessagingSession();

      // get the problematic command message from the exception
      Message message = e.message;

      // remove the problematic message from the queue
      dbMessageSessionImpl.receiveByIdNoWait(message.getId());
      
      message = Message.createCopy(message);

      // update the message with the stack trace
      StringWriter sw = new StringWriter();
      e.printStackTrace(new PrintWriter(sw));
      message.setException(sw.toString());

      // update the message with the jbpm-error-queue destination
      message.setDestination(errorDestination);
      
      // resend
      messageSession.save(message);

    } finally {
      jbpmContext.close();
    }
  }

  static class MessageProcessingException extends Exception {
    private static final long serialVersionUID = 1L;
    Message message;
    public MessageProcessingException(Message message, Throwable cause) {
      super("message "+message+"' couldn't be processed", cause);
      this.message = message;
    }
  }

  private static Log log = LogFactory.getLog(CommandExecutorThread.class);
}