public InvocationResponse invoke()

in modules/core/src/main/java/org/apache/sandesha2/handlers/SandeshaInHandler.java [62:193]
92 lines of code
44 McCabe index (conditional complexity)

	public InvocationResponse invoke(MessageContext msgCtx) throws AxisFault {
		
		if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
			log.debug("Enter: SandeshaInHandler::invoke, " + msgCtx.getEnvelope().getHeader());

		InvocationResponse returnValue = InvocationResponse.CONTINUE;
		
		ConfigurationContext context = msgCtx.getConfigurationContext();
		if (context == null) {
			String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.configContextNotSet);
			if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) log.debug(message);
			throw new AxisFault(message);
		}

		String DONE = (String) msgCtx.getProperty(Sandesha2Constants.APPLICATION_PROCESSING_DONE);
		if (null != DONE && Sandesha2Constants.VALUE_TRUE.equals(DONE)) {
			if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
				log.debug("Exit: SandeshaInHandler::invoke, Application processing done " + returnValue);
			return returnValue;
		}
		
		// look at the service to see if RM is totally disabled. This allows the user to disable RM using
		// a property on the service, even when Sandesha is engaged.
		if (msgCtx.getAxisService() != null) {
			Parameter unreliableParam = msgCtx.getAxisService().getParameter(SandeshaClientConstants.UNRELIABLE_MESSAGE);
			if (null != unreliableParam && "true".equals(unreliableParam.getValue())) {
				if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) log.debug("Exit: SandeshaInHandler::invoke, Service has disabled RM " + returnValue);
				return returnValue;
			}
		}
		if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) log.debug("SandeshaInHandler::invoke Continuing beyond basic checks");

		Transaction transaction = null;

		try {
			StorageManager storageManager = SandeshaUtil.getSandeshaStorageManager(context, context.getAxisConfiguration());

			AxisService axisService = msgCtx.getAxisService();
			if (axisService == null) {
				String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.axisServiceIsNull);
				if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) log.debug(message);
				throw new AxisFault(message);
			}

			RMMsgContext rmMsgCtx = null;
      
			if (msgCtx.getProperty(Sandesha2Constants.MessageContextProperties.RM_MESSAGE_CONTEXT) != null)
				rmMsgCtx = (RMMsgContext)msgCtx.getProperty(Sandesha2Constants.MessageContextProperties.RM_MESSAGE_CONTEXT);
			else
				rmMsgCtx = MsgInitializer.initializeMessage(msgCtx);

			if (rmMsgCtx.getMessageType() == Sandesha2Constants.MessageTypes.UNKNOWN) {
				// checking if policies hv been set to enforceRM.
				// If this is set and this message is not an RM message, validation
				// will fail here.

				SandeshaPolicyBean propertyBean = SandeshaUtil
						.getPropertyBean(msgCtx.getAxisOperation());
				
				if (propertyBean.isEnforceRM() && !Boolean.valueOf((String)msgCtx.getProperty(Sandesha2Constants.FORCE_ALLOW_UNRELIABLE_MSG))) {
					String message = SandeshaMessageHelper.getMessage(
							SandeshaMessageKeys.rmEnforceFailure, msgCtx.getMessageID());
					log.warn(message);
					throw new SandeshaException(message);
				}
			}
			
			transaction = storageManager.getTransaction();

			// Process Ack headers in the message
			AcknowledgementProcessor ackProcessor = new AcknowledgementProcessor();
			ackProcessor.processAckHeaders(rmMsgCtx);

			// commit the current transaction
			if(transaction != null && transaction.isActive()) transaction.commit();
			transaction = storageManager.getTransaction();

			// Process Ack Request headers in the message
			AckRequestedProcessor reqProcessor = new AckRequestedProcessor();
			if(reqProcessor.processAckRequestedHeaders(rmMsgCtx)){
				returnValue = InvocationResponse.SUSPEND;
				//msgCtx.setProperty(RequestResponseTransport.HOLD_RESPONSE, Boolean.TRUE);
			}
			
			// Process MessagePending headers
			MessagePendingProcessor pendingProcessor = new MessagePendingProcessor();
			pendingProcessor.processMessagePendingHeaders(rmMsgCtx);

			// commit the current transaction
			if(transaction != null && transaction.isActive()) transaction.commit();
			transaction = storageManager.getTransaction();

			// Process the Sequence header, if there is one
			SequenceProcessor seqProcessor = new SequenceProcessor();
			returnValue = seqProcessor.processSequenceHeader(rmMsgCtx, transaction);

			// commit the current transaction
			if(transaction != null && transaction.isActive()) transaction.commit();
			transaction = null;
			
		} catch (Exception e) {
			if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) 
				log.debug("SandeshaInHandler::invoke Exception caught during processInMessage", e);
			// message should not be sent in a exception situation.
			msgCtx.pause();
			returnValue = InvocationResponse.SUSPEND;
			
			// Rethrow the original exception if it is an AxisFault
			if (e instanceof AxisFault)
				throw (AxisFault)e;
			
			String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.inMsgError, e.toString());
			throw new AxisFault(message, e);
		} 
		finally {
			if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled()) log.debug("SandeshaInHandler::invoke Doing final processing");
			if (transaction != null && transaction.isActive()) {
				try {
					transaction.rollback();
					transaction = null;
				} catch (Exception e) {
					String message = SandeshaMessageHelper.getMessage(SandeshaMessageKeys.rollbackError, e.toString());
					if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
                      log.debug(message, e);
				}
			}
		}
		
		if (LoggingControl.isAnyTracingEnabled() && log.isDebugEnabled())
			log.debug("Exit: SandeshaInHandler::invoke " + returnValue);
		return returnValue;
	}