src/org/apache/sandesha2/workers/Sender.java

/*
 * Copyright 1999-2004 The Apache Software Foundation.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 *  
 */
package org.apache.sandesha2.workers;

import java.util.Collection;
import java.util.Iterator;
import org.apache.axis2.AxisFault;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.context.ServiceContext;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.soap.SOAPEnvelope;
import org.apache.sandesha2.AcknowledgementManager;
import org.apache.sandesha2.Sandesha2ClientAPI;
import org.apache.sandesha2.Sandesha2Constants;
import org.apache.sandesha2.RMMsgContext;
import org.apache.sandesha2.SandeshaException;
import org.apache.sandesha2.TerminateManager;
import org.apache.sandesha2.storage.StorageManager;
import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr;
import org.apache.sandesha2.storage.beans.SenderBean;
import org.apache.sandesha2.util.MessageRetransmissionAdjuster;
import org.apache.sandesha2.util.MsgInitializer;
import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.wsrm.TerminateSequence;

/**
 * This is responsible for sending and re-sending messages of Sandesha2. This represent a thread that keep running all
 * the time. This keep looking at the Sender table to find out any entries that should be sent.
 * 
 * @author Chamikara Jayalath <chamikaramj@gmail.com>
 */

public class Sender extends Thread {

	private boolean senderStarted = false;

	private ConfigurationContext context = null;

	public synchronized void stopSender() {
		senderStarted = false;
	}

	public synchronized boolean isSenderStarted() {
		return senderStarted;
	}

	public void run() {

		while (senderStarted) {
			try {
				if (context == null)
					throw new SandeshaException(
							"Can't continue the Sender. Context is null");

				StorageManager storageManager = SandeshaUtil
						.getSandeshaStorageManager(context);

				SenderBeanMgr mgr = storageManager
						.getRetransmitterBeanMgr();
				Collection coll = mgr.findMsgsToSend();
				Iterator iter = coll.iterator();

				while (iter.hasNext()) {

					SenderBean bean = (SenderBean) iter.next();
					String key = (String) bean.getKey();
					MessageContext msgCtx = SandeshaUtil
							.getStoredMessageContext(key);

					try {
						RMMsgContext rmMsgCtx = MsgInitializer
								.initializeMessage(msgCtx);

						updateMessage(msgCtx);

						ServiceContext serviceContext = msgCtx
								.getServiceContext();
						Object debug = null;
						if (serviceContext != null) {
							debug = msgCtx
									.getProperty(Sandesha2ClientAPI.SANDESHA_DEBUG_MODE);
							if (debug != null && "on".equals(debug)) {
								System.out
										.println("DEBUG: Sender is sending a '"
												+ SandeshaUtil
														.getMessageTypeString(rmMsgCtx
																.getMessageType())
												+ "' message.");
							}
						}

						if (rmMsgCtx.getMessageType() == Sandesha2Constants.MessageTypes.APPLICATION) {
							//piggybacking if an ack if available for the same
							// sequence.
							AcknowledgementManager
									.piggybackAckIfPresent(rmMsgCtx);
						}

						try {
							new AxisEngine(context).send(msgCtx);
						} catch (Exception e) {
							//Exception is sending. retry later
							System.out
									.println("Exception thrown in sending...");
							e.printStackTrace();
						}

						MessageRetransmissionAdjuster retransmitterAdjuster = new MessageRetransmissionAdjuster();
						retransmitterAdjuster.adjustRetransmittion(bean);

						mgr.update(bean);

						if (!msgCtx.isServerSide())
							checkForSyncResponses(msgCtx);
						
						
						if (rmMsgCtx.getMessageType()==Sandesha2Constants.MessageTypes.TERMINATE_SEQ) {
							//terminate sending side.
							TerminateSequence terminateSequence = (TerminateSequence) rmMsgCtx.getMessagePart(Sandesha2Constants.MessageParts.TERMINATE_SEQ);
							String sequenceID = terminateSequence.getIdentifier().getIdentifier();
							ConfigurationContext configContext = msgCtx.getConfigurationContext();
							
							TerminateManager.terminateSendingSide(configContext,sequenceID);
						}

					} catch (AxisFault e1) {
						e1.printStackTrace();
					} catch (Exception e3) {
						e3.printStackTrace();
					}

					//changing the values of the sent bean.
					//bean.setLastSentTime(System.currentTimeMillis());
					//bean.setSentCount(bean.getSentCount() + 1);

					//update if resend=true otherwise delete. (reSend=false
					// means
					// send only once).
					if (bean.isReSend())
						mgr.update(bean);
					else
						mgr.delete(bean.getMessageId());

				}
			} catch (SandeshaException e) {
				e.printStackTrace();
				return;
			}

			try {
				Thread.sleep(2000);
			} catch (InterruptedException e1) {
				//e1.printStackTrace();
				System.out.println("Sender was interupted...");
				e1.printStackTrace();
				System.out.println("End printing Interrupt...");
			}
		}

	}

	private boolean isResponseExpected(RMMsgContext rmMsgCtx) {
		boolean responseExpected = false;

		if (rmMsgCtx.getMessageType() == Sandesha2Constants.MessageTypes.CREATE_SEQ) {
			responseExpected = true;
		}
		if (rmMsgCtx.getMessageType() == Sandesha2Constants.MessageTypes.APPLICATION) {
			//a ack may arrive. (not a application response)
			if (rmMsgCtx.getMessageContext().getAxisOperation()
					.getMessageExchangePattern().equals(
							org.apache.wsdl.WSDLConstants.MEP_URI_IN_OUT)) {
				responseExpected = true;
			}
		}

		return true;
	}

	public void start(ConfigurationContext context) {
		senderStarted = true;
		this.context = context;
		super.start();
	}

	private void updateMessage(MessageContext msgCtx1) throws SandeshaException {
		try {
			RMMsgContext rmMsgCtx1 = MsgInitializer.initializeMessage(msgCtx1);
			rmMsgCtx1.addSOAPEnvelope();

		} catch (AxisFault e) {
			throw new SandeshaException("Exception in updating contexts");
		}

	}

	private void checkForSyncResponses(MessageContext msgCtx) throws AxisFault {

		boolean responsePresent = (msgCtx
				.getProperty(MessageContext.TRANSPORT_IN) != null);

		if (responsePresent) {
			//create the response
			MessageContext response = new MessageContext(msgCtx
					.getConfigurationContext(), msgCtx.getSessionContext(), msgCtx
					.getTransportIn(), msgCtx.getTransportOut());
			response.setProperty(MessageContext.TRANSPORT_IN, msgCtx
					.getProperty(MessageContext.TRANSPORT_IN));

			response.setServerSide(false);

			//If request is REST we assume the response is REST, so set the
			// variable
			response.setDoingREST(msgCtx.isDoingREST());
			response
					.setServiceGroupContextId(msgCtx.getServiceGroupContextId());
			response.setServiceGroupContext(msgCtx.getServiceGroupContext());
			response.setServiceContext(msgCtx.getServiceContext());
			response.setAxisService(msgCtx.getAxisService());
			response.setAxisServiceGroup(msgCtx.getAxisServiceGroup());

			//setting the in-flow.
			//ArrayList inPhaseHandlers =
			// response.getAxisOperation().getRemainingPhasesInFlow();
			/*
			 * if (inPhaseHandlers==null || inPhaseHandlers.isEmpty()) {
			 * ArrayList phases =
			 * msgCtx.getSystemContext().getAxisConfiguration().getInPhasesUptoAndIncludingPostDispatch();
			 * response.getAxisOperation().setRemainingPhasesInFlow(phases); }
			 */

			//Changed following from TransportUtils to SandeshaUtil since op.
			// context is anavailable.
			SOAPEnvelope resenvelope = null;
			try {
				resenvelope = SandeshaUtil.createSOAPMessage(response, msgCtx
						.getEnvelope().getNamespace().getName());
			} catch (AxisFault e) {
				//TODO: change to log.debug
				e.printStackTrace();
			}

			if (resenvelope != null) {
				AxisEngine engine = new AxisEngine(msgCtx.getConfigurationContext());
				response.setEnvelope(resenvelope);
				engine.receive(response);
			}
		}
	}

}