/*
* This file is part of the WfMOpen project.
* Copyright (C) 2001-2003 Danet GmbH (www.danet.de), GS-AN.
* All rights reserved.
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
* $Id: Sender.java 3205 2009-09-27 17:13:47Z mlipp $
*
* $Log$
* Revision 1.2 2006/09/29 12:32:13 drmlipp
* Consistently using WfMOpen as projct name now.
*
* Revision 1.1.1.2 2004/08/18 15:17:38 drmlipp
* Update to 1.2
*
* Revision 1.18 2004/07/15 08:35:34 lipp
* Removed direct invocability again, as there seem to be problem
* allocating the topic connection. Will be investigated later.
*
* Revision 1.17 2004/07/14 14:54:37 lipp
* Restrictions for being direct invocable relaxed a bit.
*
* Revision 1.16 2004/07/02 13:43:37 lipp
* Fixed JMS usage.
*
* Revision 1.15 2004/04/12 19:33:52 lipp
* Clarified application invocation interface.
*
* Revision 1.14 2004/03/31 19:36:20 lipp
* Completed implementation of Activity.abandon(String).
*
* Revision 1.13 2004/03/21 12:37:55 lipp
* Fixed bug with self-sending.
*
* Revision 1.12 2004/03/20 21:08:44 lipp
* Added access to requesting processes' channels.
*
* Revision 1.11 2004/02/25 12:13:08 lipp
* Fixed silly bug.
*
* Revision 1.10 2004/02/25 12:06:06 lipp
* Added some debugging info.
*
* Revision 1.9 2004/02/19 15:52:12 schlue
* Changed default for local delivery.
*
* Revision 1.8 2004/02/13 10:01:34 lipp
* Changed result type for result provider to Map which is more
* appropriate.
*
* Revision 1.7 2004/02/13 08:25:29 lipp
* Changed channel message data type to Map which is more appropriate.
*
* Revision 1.6 2004/02/06 13:37:35 lipp
* Added channel close notification.
*
* Revision 1.5 2004/01/28 16:11:38 lipp
* Re-implementation of chabacc, Sender working.
*
* Revision 1.4 2003/10/08 15:15:35 lipp
* Updated channel handling.
*
* Revision 1.3 2003/10/07 15:49:53 lipp
* Completed transmission sequence to activity.
*
* Revision 1.2 2003/10/06 15:25:56 lipp
* Using proper complete procedure.
*
* Revision 1.1 2003/10/05 15:40:18 lipp
* Continuing chabacc implementation.
*
*/
package de.danet.an.workflow.tools.chabacc;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.rmi.RemoteException;
import de.danet.an.util.CollectionsUtil;
import de.danet.an.util.EJBUtil;
import de.danet.an.util.ResourceNotAvailableException;
import de.danet.an.workflow.omgcore.InvalidDataException;
import de.danet.an.workflow.api.Activity;
import de.danet.an.workflow.api.ActivityUniqueKey;
import de.danet.an.workflow.api.FormalParameter;
import de.danet.an.workflow.apix.ExtProcess;
import de.danet.an.workflow.spis.aii.ApplicationNotStoppedException;
import de.danet.an.workflow.spis.aii.CannotExecuteException;
import de.danet.an.workflow.spis.aii.ResultProvider;
import de.danet.an.workflow.spis.aii.ToolAgent;
/**
* This class provides a tool that sends messages to a channel. The
* first argument must be of type string and identifies the
* channel. The second and subsequent arguments may be of any type and
* will be send to the channel unmodified.
*
* @author <a href="mailto:lipp@danet.de">Michael Lipp</a>
* @version $Revision: 3205 $
*/
public class Sender
implements ToolAgent, ResultProvider {
private static final org.apache.commons.logging.Log logger
= org.apache.commons.logging.LogFactory.getLog(Sender.class);
private boolean localDelivery = false;
private Integer originatorIndex = null;
/**
* Creates an instance of <code>Sender</code>
* with all attributes initialized to default values.
*/
public Sender () {
}
/**
* By default, messages are delivered to clients listening on the
* channel only. If messages should also be delivered to receiver
* tools, this property must be set to <code>true</code>. Note
* that in this case the channel cannot be used for bidirectional
* communication.
*
* @param enabled a string denoting a boolean value. Any string
* that, ignoring case, equals "true" will enable local delivery.
*/
public void setLocalDelivery (String enabled) {
localDelivery = (new Boolean(enabled)).booleanValue();
}
/**
* Normally, the Sender sends data on a channel of the process it
* is invoked in. It is, however, also possible to send the data
* in a channel of its requesters. See the user manual for
* details.
* @param index used to determine the originating process, see
* user manual
*/
public void setOriginatorIndex (String index) {
originatorIndex = new Integer (index);
}
// Implementation of de.danet.an.workflow.spis.aii.ToolAgent
/* Comment copied from interface. */
public void invoke
(Activity activity, FormalParameter[] formalParameter, Map map)
throws CannotExecuteException, RemoteException {
ActivityUniqueKey auk = null;
String channelName = null;
try {
auk = activity.uniqueKey();
channelName = (String)map.get(formalParameter[0].id());
if (logger.isDebugEnabled ()) {
logger.debug (auk + " sending data to " + channelName);
}
Map data = new HashMap ();
for (int i = 1; i < formalParameter.length; i++) {
String fid = formalParameter[i].id();
data.put (fid, map.get (fid));
}
if (logger.isDebugEnabled()) {
logger.debug("Sending " + CollectionsUtil.toString(data));
}
ExtProcess proc
= (originatorIndex == null
? (ExtProcess)activity.container()
: getOriginator (activity));
proc.broadcastChannelMessage(channelName, data);
if (localDelivery) {
// Loop back message to process for tool-to-tool messages
proc.deliverChannelMessage (channelName, data);
}
if (logger.isDebugEnabled ()) {
logger.debug (auk + " sent data to "
+ map.get(formalParameter[0].id()));
}
} catch (ResourceNotAvailableException e) {
logger.error (e.getMessage(), e);
throw new CannotExecuteException (e.getMessage());
} catch (InvalidDataException e) {
logger.error (e.getMessage(), e);
throw new CannotExecuteException (e.getMessage());
}
}
private ExtProcess getOriginator (Activity act)
throws CannotExecuteException, RemoteException {
ExtProcess proc = (ExtProcess)act.container();
if (originatorIndex.intValue() > 0) {
for (int i = originatorIndex.intValue(); i > 0; i--) {
proc = (ExtProcess)proc.requestingProcess();
if (proc == null) {
throw new CannotExecuteException
(act + " does not have " + originatorIndex
+ " requesting anchestors");
}
}
} else {
List procs = new ArrayList ();
while (true) {
procs.add (0, proc);
proc = (ExtProcess)proc.requestingProcess();
if (proc == null) {
break;
}
}
try {
proc = (ExtProcess)procs.get (-originatorIndex.intValue());
} catch (IndexOutOfBoundsException e) {
throw new CannotExecuteException
("Subflow depth from root of " + act
+ " is less than " + -originatorIndex.intValue());
}
}
return proc;
}
/* Comment copied from interface. */
public void terminate(Activity activity)
throws ApplicationNotStoppedException {
}
/* Comment copied from interface. */
public Object result () {
return null;
}
}