<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you 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.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.axiom.om.impl.builder.StAXBuilder;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.AddressingConstants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.context.OperationContext;
import org.apache.axis2.context.OperationContextFactory;
import org.apache.axis2.context.ServiceContext;
import org.apache.axis2.description.AxisOperation;
import org.apache.axis2.description.OutOnlyAxisOperation;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.engine.Handler.InvocationResponse;
import org.apache.axis2.transport.RequestResponseTransport;
import org.apache.axis2.transport.TransportUtils;
import org.apache.axis2.transport.RequestResponseTransport.RequestResponseTransportStatus;
import org.apache.axis2.transport.http.HTTPConstants;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.sandesha2.RMMsgContext;
import org.apache.sandesha2.Sandesha2Constants;
import org.apache.sandesha2.SandeshaException;
import org.apache.sandesha2.i18n.SandeshaMessageHelper;
import org.apache.sandesha2.i18n.SandeshaMessageKeys;
import org.apache.sandesha2.storage.SandeshaStorageException;
import org.apache.sandesha2.storage.StorageManager;
import org.apache.sandesha2.storage.Transaction;
import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr;
import org.apache.sandesha2.storage.beans.RMDBean;
import org.apache.sandesha2.storage.beans.RMSBean;
import org.apache.sandesha2.storage.beans.SenderBean;
import org.apache.sandesha2.util.AcknowledgementManager;
import org.apache.sandesha2.util.MessageRetransmissionAdjuster;
import org.apache.sandesha2.util.MsgInitializer;
import org.apache.sandesha2.util.RMMsgCreator;
import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.util.SpecSpecificConstants;
import org.apache.sandesha2.util.TerminateManager;
import org.apache.sandesha2.wsrm.AckRequested;
import org.apache.sandesha2.wsrm.CloseSequence;
import org.apache.sandesha2.wsrm.Identifier;
import org.apache.sandesha2.wsrm.Sequence;
import org.apache.sandesha2.wsrm.TerminateSequence;

public class SenderWorker extends SandeshaWorker implements Runnable {

  private static final Log log = LogFactory.getLog(SenderWorker.class);

	private ConfigurationContext configurationContext = null;
	private SenderBean senderBean = null;
	private RMMsgContext messageToSend = null;
	private String rmVersion = null;

	public SenderWorker (ConfigurationContext configurationContext, SenderBean senderBean, String rmVersion) {
		this.configurationContext = configurationContext;
		this.senderBean = senderBean;
		this.rmVersion = rmVersion; 
	}
	
	public void setMessage(RMMsgContext msg) {
		this.messageToSend = msg;
	}
	
	public void run () {
		
		if (log.isDebugEnabled())
			log.debug("Enter: SenderWorker::run");
		
	    // If we are not the holder of the correct lock, then we have to stop
	    if(lock != null && !lock.ownsLock(workId, this)) {
	      if (log.isDebugEnabled()) log.debug("Exit: SenderWorker::run, another worker holds the lock");
	      return;
	    }

		Transaction transaction = null;
		
		try {
			StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configurationContext, configurationContext.getAxisConfiguration());
			SenderBeanMgr senderBeanMgr = storageManager.getSenderBeanMgr();
			
			transaction = storageManager.getTransaction();

			String key = senderBean.getMessageContextRefKey();
			MessageContext msgCtx = null;
			RMMsgContext   rmMsgCtx = null;
			if(messageToSend != null) {
				msgCtx = messageToSend.getMessageContext();
				rmMsgCtx = messageToSend;
			} else {
				msgCtx = storageManager.retrieveMessageContext(key, configurationContext);
      
				if (msgCtx == null) {
					// This sender bean has already been processed
					
					if(transaction != null && transaction.isActive()) transaction.commit();
					transaction = null;

					return;
				}
      
				rmMsgCtx = MsgInitializer.initializeMessage(msgCtx);
			}

			// sender will not send the message if following property is
			// set and not true.
			// But it will set if it is not set (null)

			// This is used to make sure that the mesage get passed the
			// Sandesha2TransportSender.

			String qualifiedForSending = (String) msgCtx.getProperty(Sandesha2Constants.QUALIFIED_FOR_SENDING);
			if (qualifiedForSending != null && !qualifiedForSending.equals(Sandesha2Constants.VALUE_TRUE)) {
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, !qualified for sending");
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;
			}

			if (msgCtx == null) {
				if (log.isDebugEnabled())
					log.debug(SandeshaMessageHelper.getMessage(SandeshaMessageKeys.sendHasUnavailableMsgEntry));
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;			
			}

			// operation is the lowest level Sandesha2 should be attached
			ArrayList msgsNotToSend = SandeshaUtil.getPropertyBean(msgCtx.getAxisOperation()).getMsgTypesToDrop();

			if (msgsNotToSend != null && msgsNotToSend.contains(new Integer(rmMsgCtx.getMessageType()))) {
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, message type to be dropped " + rmMsgCtx.getMessageType());
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;	
			}

			// If we are sending to the anonymous URI then we _must_ have a transport waiting,
			// or the message can't go anywhere. If there is nothing here then we leave the
			// message in the sender queue, and a MakeConnection (or a retransmitted request)
			// will hopefully pick it up soon.
			Boolean makeConnection = (Boolean) msgCtx.getProperty(Sandesha2Constants.MAKE_CONNECTION_RESPONSE);
			EndpointReference toEPR = msgCtx.getTo();

			MessageContext inMsg = null;
			OperationContext op = msgCtx.getOperationContext();
			
			RequestResponseTransport t = (RequestResponseTransport) msgCtx.getProperty(RequestResponseTransport.TRANSPORT_CONTROL);
			
			if (t==null) {
				if (op != null)
					inMsg = op.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);
				if (inMsg != null)
					t = (RequestResponseTransport) inMsg.getProperty(RequestResponseTransport.TRANSPORT_CONTROL);
			}

			// If we are anonymous, and this is not a makeConnection, then we must have a transport waiting
			if((toEPR==null || toEPR.hasAnonymousAddress()) &&
			   (makeConnection == null || !makeConnection.booleanValue()) &&
			   (t == null || !t.getStatus().equals(RequestResponseTransportStatus.WAITING))) {
				
				// Mark this sender bean so that we know that the transport is unavailable, if the
				// bean is still stored.
				SenderBean bean = senderBeanMgr.retrieve(senderBean.getMessageID());
				if(bean != null && bean.isTransportAvailable()) {
					bean.setTransportAvailable(false);
					senderBeanMgr.update(bean);
				}
				
				// Commit the update
				if(transaction != null && transaction.isActive()) transaction.commit();
				transaction = null;
				
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, no response transport for anonymous message");
				return;
			}
			
			int messageType = senderBean.getMessageType();
			
			if (isAckPiggybackableMsgType(messageType)) {
				// Commit the update
				if(transaction != null && transaction.isActive()) transaction.commit();
				transaction = storageManager.getTransaction();

				// Piggyback ack messages based on the 'To' address of the message
				AcknowledgementManager.piggybackAcksIfPresent(rmMsgCtx, storageManager);
			}
			if (transaction != null && transaction.isActive()) 
				transaction.commit();			
			
			transaction = storageManager.getTransaction();

			
			if (transaction != null && transaction.isActive()) 
				transaction.commit();			
			
			transaction = storageManager.getTransaction();
			
			senderBean = updateMessage(rmMsgCtx,senderBean,storageManager);

			if (senderBean == null) { 
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, !continueSending");
				
				if(transaction != null && transaction.isActive()) {
					transaction.rollback();
					transaction = null;
				}
				
				return;
			}

			// Although not actually sent yet, update the send count to indicate an attempt
			if (senderBean.isReSend()) {
				senderBeanMgr.update(senderBean);
			}

			// sending the message
			boolean successfullySent = false;
			
			//try to redecorate the EPR if necessary
			if (log.isDebugEnabled())
				log.debug("Redecorate EPR : " + msgCtx.getEnvelope().getHeader());
			EndpointReference replyToEPR = msgCtx.getReplyTo();
			if(replyToEPR!=null){
				replyToEPR = SandeshaUtil.getEPRDecorator(msgCtx.getConfigurationContext()).decorateEndpointReference(replyToEPR);
				msgCtx.setReplyTo(replyToEPR); 
			}
			
			// have to commit the transaction before sending. This may
			// get changed when WS-AT is available.
			if(transaction != null) {
				transaction.commit();
				transaction = null;
			}

			msgCtx.getOptions().setTimeOutInMilliSeconds(1000000);
			
			boolean processResponseForFaults = false ;
			try {
				InvocationResponse response = InvocationResponse.CONTINUE;
				
		        if(storageManager.requiresMessageSerialization()) {
					if(msgCtx.isPaused()) {
						if (log.isDebugEnabled())
							log.debug("Resuming a send for message : " + msgCtx.getEnvelope().getHeader());
						msgCtx.setPaused(false);<commit_msg>Performance: Reduce the EPR Decorator cost by not modifying application message ReplyTo EPRs Thanks to <NAME> for help with this patch<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you 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.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

import org.apache.axiom.om.impl.builder.StAXBuilder;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.AddressingConstants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.context.OperationContext;
import org.apache.axis2.context.OperationContextFactory;
import org.apache.axis2.context.ServiceContext;
import org.apache.axis2.description.AxisOperation;
import org.apache.axis2.description.OutOnlyAxisOperation;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.engine.Handler.InvocationResponse;
import org.apache.axis2.transport.RequestResponseTransport;
import org.apache.axis2.transport.TransportUtils;
import org.apache.axis2.transport.RequestResponseTransport.RequestResponseTransportStatus;
import org.apache.axis2.transport.http.HTTPConstants;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.sandesha2.RMMsgContext;
import org.apache.sandesha2.Sandesha2Constants;
import org.apache.sandesha2.SandeshaException;
import org.apache.sandesha2.i18n.SandeshaMessageHelper;
import org.apache.sandesha2.i18n.SandeshaMessageKeys;
import org.apache.sandesha2.storage.SandeshaStorageException;
import org.apache.sandesha2.storage.StorageManager;
import org.apache.sandesha2.storage.Transaction;
import org.apache.sandesha2.storage.beanmanagers.SenderBeanMgr;
import org.apache.sandesha2.storage.beans.RMDBean;
import org.apache.sandesha2.storage.beans.RMSBean;
import org.apache.sandesha2.storage.beans.SenderBean;
import org.apache.sandesha2.util.AcknowledgementManager;
import org.apache.sandesha2.util.MessageRetransmissionAdjuster;
import org.apache.sandesha2.util.MsgInitializer;
import org.apache.sandesha2.util.RMMsgCreator;
import org.apache.sandesha2.util.SandeshaUtil;
import org.apache.sandesha2.util.SpecSpecificConstants;
import org.apache.sandesha2.util.TerminateManager;
import org.apache.sandesha2.wsrm.AckRequested;
import org.apache.sandesha2.wsrm.CloseSequence;
import org.apache.sandesha2.wsrm.Identifier;
import org.apache.sandesha2.wsrm.Sequence;
import org.apache.sandesha2.wsrm.TerminateSequence;

public class SenderWorker extends SandeshaWorker implements Runnable {

  private static final Log log = LogFactory.getLog(SenderWorker.class);

	private ConfigurationContext configurationContext = null;
	private SenderBean senderBean = null;
	private RMMsgContext messageToSend = null;
	private String rmVersion = null;

	public SenderWorker (ConfigurationContext configurationContext, SenderBean senderBean, String rmVersion) {
		this.configurationContext = configurationContext;
		this.senderBean = senderBean;
		this.rmVersion = rmVersion; 
	}
	
	public void setMessage(RMMsgContext msg) {
		this.messageToSend = msg;
	}
	
	public void run () {
		
		if (log.isDebugEnabled())
			log.debug("Enter: SenderWorker::run");
		
	    // If we are not the holder of the correct lock, then we have to stop
	    if(lock != null && !lock.ownsLock(workId, this)) {
	      if (log.isDebugEnabled()) log.debug("Exit: SenderWorker::run, another worker holds the lock");
	      return;
	    }

		Transaction transaction = null;
		
		try {
			StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(configurationContext, configurationContext.getAxisConfiguration());
			SenderBeanMgr senderBeanMgr = storageManager.getSenderBeanMgr();
			
			transaction = storageManager.getTransaction();

			String key = senderBean.getMessageContextRefKey();
			MessageContext msgCtx = null;
			RMMsgContext   rmMsgCtx = null;
			if(messageToSend != null) {
				msgCtx = messageToSend.getMessageContext();
				rmMsgCtx = messageToSend;
			} else {
				msgCtx = storageManager.retrieveMessageContext(key, configurationContext);
      
				if (msgCtx == null) {
					// This sender bean has already been processed
					
					if(transaction != null && transaction.isActive()) transaction.commit();
					transaction = null;

					return;
				}
      
				rmMsgCtx = MsgInitializer.initializeMessage(msgCtx);
			}

			// sender will not send the message if following property is
			// set and not true.
			// But it will set if it is not set (null)

			// This is used to make sure that the mesage get passed the
			// Sandesha2TransportSender.

			String qualifiedForSending = (String) msgCtx.getProperty(Sandesha2Constants.QUALIFIED_FOR_SENDING);
			if (qualifiedForSending != null && !qualifiedForSending.equals(Sandesha2Constants.VALUE_TRUE)) {
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, !qualified for sending");
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;
			}

			if (msgCtx == null) {
				if (log.isDebugEnabled())
					log.debug(SandeshaMessageHelper.getMessage(SandeshaMessageKeys.sendHasUnavailableMsgEntry));
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;			
			}

			// operation is the lowest level Sandesha2 should be attached
			ArrayList msgsNotToSend = SandeshaUtil.getPropertyBean(msgCtx.getAxisOperation()).getMsgTypesToDrop();

			if (msgsNotToSend != null && msgsNotToSend.contains(new Integer(rmMsgCtx.getMessageType()))) {
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, message type to be dropped " + rmMsgCtx.getMessageType());
				
				if(transaction != null && transaction.isActive()) {
					transaction.commit();
					transaction = null;
				}
				
				return;	
			}

			// If we are sending to the anonymous URI then we _must_ have a transport waiting,
			// or the message can't go anywhere. If there is nothing here then we leave the
			// message in the sender queue, and a MakeConnection (or a retransmitted request)
			// will hopefully pick it up soon.
			Boolean makeConnection = (Boolean) msgCtx.getProperty(Sandesha2Constants.MAKE_CONNECTION_RESPONSE);
			EndpointReference toEPR = msgCtx.getTo();

			MessageContext inMsg = null;
			OperationContext op = msgCtx.getOperationContext();
			
			RequestResponseTransport t = (RequestResponseTransport) msgCtx.getProperty(RequestResponseTransport.TRANSPORT_CONTROL);
			
			if (t==null) {
				if (op != null)
					inMsg = op.getMessageContext(WSDLConstants.MESSAGE_LABEL_IN_VALUE);
				if (inMsg != null)
					t = (RequestResponseTransport) inMsg.getProperty(RequestResponseTransport.TRANSPORT_CONTROL);
			}

			// If we are anonymous, and this is not a makeConnection, then we must have a transport waiting
			if((toEPR==null || toEPR.hasAnonymousAddress()) &&
			   (makeConnection == null || !makeConnection.booleanValue()) &&
			   (t == null || !t.getStatus().equals(RequestResponseTransportStatus.WAITING))) {
				
				// Mark this sender bean so that we know that the transport is unavailable, if the
				// bean is still stored.
				SenderBean bean = senderBeanMgr.retrieve(senderBean.getMessageID());
				if(bean != null && bean.isTransportAvailable()) {
					bean.setTransportAvailable(false);
					senderBeanMgr.update(bean);
				}
				
				// Commit the update
				if(transaction != null && transaction.isActive()) transaction.commit();
				transaction = null;
				
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, no response transport for anonymous message");
				return;
			}
			
			int messageType = senderBean.getMessageType();
			
			if (isAckPiggybackableMsgType(messageType)) {
				// Commit the update
				if(transaction != null && transaction.isActive()) transaction.commit();
				transaction = storageManager.getTransaction();

				// Piggyback ack messages based on the 'To' address of the message
				AcknowledgementManager.piggybackAcksIfPresent(rmMsgCtx, storageManager);
			}
			if (transaction != null && transaction.isActive()) 
				transaction.commit();			
			
			transaction = storageManager.getTransaction();

			
			if (transaction != null && transaction.isActive()) 
				transaction.commit();			
			
			transaction = storageManager.getTransaction();
			
			senderBean = updateMessage(rmMsgCtx,senderBean,storageManager);

			if (senderBean == null) { 
				if (log.isDebugEnabled())
					log.debug("Exit: SenderWorker::run, !continueSending");
				
				if(transaction != null && transaction.isActive()) {
					transaction.rollback();
					transaction = null;
				}
				
				return;
			}

			// Although not actually sent yet, update the send count to indicate an attempt
			if (senderBean.isReSend()) {
				senderBeanMgr.update(senderBean);
			}

			// sending the message
			boolean successfullySent = false;
			
			// No need to redecorate application messages ... only for rm protocol messages
			if(Sandesha2Constants.MessageTypes.APPLICATION!=senderBean.getMessageType()){
				//try to redecorate the EPR if necessary
				if (log.isDebugEnabled())
					log.debug("Redecorate EPR : " + msgCtx.getEnvelope().getHeader());
				EndpointReference replyToEPR = msgCtx.getReplyTo();
				if(replyToEPR!=null){
					replyToEPR = SandeshaUtil.getEPRDecorator(msgCtx.getConfigurationContext()).decorateEndpointReference(replyToEPR);
					msgCtx.setReplyTo(replyToEPR); 
				}
			}
			// have to commit the transaction before sending. This may
			// get changed when WS-AT is available.
			if(transaction != null) {
				transaction.commit();
				transaction = null;
			}

			msgCtx.getOptions().setTimeOutInMilliSeconds(1000000);
			
			boolean processResponseForFaults = false ;
			try {
				InvocationResponse response = InvocationResponse.CONTINUE;
				
		        if(storageManager.requiresMessageSerialization()) {
					if(msgCtx.isPaused()) {
						if (log.isDebugEnabled())
							log.debug("Resuming a send for message : " + msgCtx.getEnvelope().getHeader());
						msgCtx.setPaused(false);
{'fraction_non_alphanumeric': 0.09899405578417925, 'fraction_numerical': 0.005898491083676269, 'mean_word_length': 4.161907009676658, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1221023', 'n_tokens_mistral': 7343, 'n_tokens_neox': 6229, 'n_words': 1517}