<commit_before>package gov.nih.nci.cagrid.caxchange.service;

import gov.nih.nci.caXchange.CaxchangeConstants;
import gov.nih.nci.caXchange.CaxchangeErrors;
import gov.nih.nci.cagrid.caxchange.context.service.globus.resource.CaXchangeResponseServiceResource;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeConnectedListeners;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeResponseExceptionListener;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeResponseListener;

import gov.nih.nci.cagrid.caxchange.listener.ResponseHandler;

import gov.nih.nci.caxchange.ResponseMessage;

import java.io.StringWriter;

import java.rmi.RemoteException;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;

import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import javax.xml.namespace.QName;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.axis.encoding.SerializationContext;
import org.apache.axis.encoding.ser.BeanSerializer;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import org.globus.wsrf.ResourceKey;
import org.globus.wsrf.security.SecurityManager;

/**
 * TODO:I am the service side implementation class.  IMPLEMENT AND DOCUMENT ME
 *
 * @created by Introduce Toolkit version 1.1
 *
 */
public class CaXchangeRequestProcessorImpl extends CaXchangeRequestProcessorImplBase {
    protected static String brokerURL=null; //"tcp://localhost:61618";
    protected static String destinationName = "caxchangeInboundQueue";
    protected static String replyDestinationName = "caxchangeOutboundQueue";
    protected static Destination destination = null;
    protected static Destination replyDestination = null;
    protected static Logger logger = LogManager.getLogger(CaXchangeRequestProcessorImpl.class.getName());
    protected static ConnectionFactory connectionFactory = null;
    protected static Connection connection = null;
    public static Map<CaxchangeResponseExceptionListener,Connection> responseListeners = null;

	public CaXchangeRequestProcessorImpl() throws RemoteException {
		super();
 	}
  /**
     * Processes a request by sending it to the caXchange inbound queue.
     *
     * @param caXchangeRequestMessage
     * @return
     * @throws RemoteException
     */
  public gov.nih.nci.cagrid.caxchange.context.stubs.types.CaXchangeResponseServiceReference processRequestAsynchronously(gov.nih.nci.caxchange.Message caXchangeRequestMessage) throws RemoteException {
	  logger.debug("request:"+caXchangeRequestMessage);
      try {
          gov.nih.nci.cagrid.caxchange.context.service.globus.resource.CaXchangeResponseServiceResourceHome ctxResourceHome = getCaXchangeResponseServiceResourceHome();
          ResourceKey resKey = ctxResourceHome.createResource();
          logger.info("Performance Request Received,"+resKey.getValue()+","+caXchangeRequestMessage.getMetadata().getMessageType()+","+new java.util.Date().getTime());
          caXchangeRequestMessage.getMetadata().setCaXchangeIdentifier(resKey.getValue().toString());
  		  String caller = SecurityManager.getManager().getCaller();
  		  if ((caller == null) || (caller.equals("<anonymous>"))) {
  			updateErrorResponse(caXchangeRequestMessage, resKey, CaxchangeErrors.PERMISSION_DENIED_FAULT, "Unable to get the identity of the caller.Caller identity:"+caller);
		  }
          caXchangeRequestMessage.getMetadata().getCredentials().setGridIdentifier(caller);
          try { //Sending the request to the caXchange inbound queue
        	 logger.info("Before sending messge "+new Date().getTime());
             sendMessage(caXchangeRequestMessage);
             logger.info("Before sending messge "+new Date().getTime());
          }catch(Exception e) {
              updateErrorResponse(caXchangeRequestMessage, resKey, CaxchangeErrors.UNKNOWN, "An error occured sending message to the caXchange hub."+e.getMessage());
          }

          if ((responseListeners==null)||(responseListeners.size()==0)) {
              registerResponseListeners();
          }
          logger.debug("Request send to ESB for key "+resKey);

          return ctxResourceHome.getResourceReference(resKey);
      }
      catch(Exception e) {
          logger.error("Error processing message request.",e);
          throw new RemoteException("Error processing message request.",e);
      }
  }
    /**
     * Send the message to the caXchange inbound JMS component.
     *
     * @param caXchangeRequestMessage
     * @throws Exception
     */
    protected void sendMessage(gov.nih.nci.caxchange.Message caXchangeRequestMessage) throws Exception {
        Session session = null;
        try  {
        	CaXchangeRequestProcessorConfiguration configuration = getConfiguration();
            CaXchangeExternalProperties properties = CaXchangeExternalProperties.getInstance();
            String inboundJmsBrokerUrl = properties.getProperty("inbound.jms.brokerURL");
            if (inboundJmsBrokerUrl==null) {
            	inboundJmsBrokerUrl = configuration.getCaXchangeInboundBrokerURL();
            }
            if (connectionFactory == null) {
                connectionFactory = new ActiveMQConnectionFactory(inboundJmsBrokerUrl);
            }
            if (destination == null) {
                destination = new ActiveMQQueue(destinationName);
            }
            if (connection == null) {
               connection = connectionFactory.createConnection();
            }
            try {
               session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }catch(JMSException connectionFailed) {
            	//Connection may have failed try to create a new connection:
            	connection = connectionFactory.createConnection();
            	session  = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }
            MessageProducer producer = session.createProducer(destination);
            RequestGeneratorHelper rgh = new RequestGeneratorHelper();
            rgh.setCaxchangeRequest(caXchangeRequestMessage);
            String req = rgh.getRequestForCaxchange();
            TextMessage message = session.createTextMessage();
            logger.debug("Sending message:"+req);
            message.setText(req);
            producer.send(message);<commit_msg>few comment lines added for implementing synchronous response<commit_after>package gov.nih.nci.cagrid.caxchange.service;

import gov.nih.nci.caXchange.CaxchangeConstants;
import gov.nih.nci.caXchange.CaxchangeErrors;
import gov.nih.nci.cagrid.caxchange.context.service.globus.resource.CaXchangeResponseServiceResource;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeConnectedListeners;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeResponseExceptionListener;
import gov.nih.nci.cagrid.caxchange.listener.CaxchangeResponseListener;

import gov.nih.nci.cagrid.caxchange.listener.ResponseHandler;

import gov.nih.nci.caxchange.ResponseMessage;

import java.io.StringWriter;

import java.rmi.RemoteException;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;

import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import javax.xml.namespace.QName;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.axis.encoding.SerializationContext;
import org.apache.axis.encoding.ser.BeanSerializer;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import org.globus.wsrf.ResourceKey;
import org.globus.wsrf.security.SecurityManager;

/**
 * TODO:I am the service side implementation class.  IMPLEMENT AND DOCUMENT ME
 *
 * @created by Introduce Toolkit version 1.1
 *
 */
public class CaXchangeRequestProcessorImpl extends CaXchangeRequestProcessorImplBase {
    protected static String brokerURL=null; //"tcp://localhost:61618";
    protected static String destinationName = "caxchangeInboundQueue";
    protected static String replyDestinationName = "caxchangeOutboundQueue";
    protected static Destination destination = null;
    protected static Destination replyDestination = null;
    protected static Logger logger = LogManager.getLogger(CaXchangeRequestProcessorImpl.class.getName());
    protected static ConnectionFactory connectionFactory = null;
    protected static Connection connection = null;
    public static Map<CaxchangeResponseExceptionListener,Connection> responseListeners = null;

	public CaXchangeRequestProcessorImpl() throws RemoteException {
		super();
 	}
  /**
     * Processes a request by sending it to the caXchange inbound queue.
     *
     * @param caXchangeRequestMessage
     * @return
     * @throws RemoteException
     */
  public gov.nih.nci.cagrid.caxchange.context.stubs.types.CaXchangeResponseServiceReference processRequestAsynchronously(gov.nih.nci.caxchange.Message caXchangeRequestMessage) throws RemoteException {
	  logger.debug("request:"+caXchangeRequestMessage);
      try {
          gov.nih.nci.cagrid.caxchange.context.service.globus.resource.CaXchangeResponseServiceResourceHome ctxResourceHome = getCaXchangeResponseServiceResourceHome();
          ResourceKey resKey = ctxResourceHome.createResource();
          logger.info("Performance Request Received,"+resKey.getValue()+","+caXchangeRequestMessage.getMetadata().getMessageType()+","+new java.util.Date().getTime());
          caXchangeRequestMessage.getMetadata().setCaXchangeIdentifier(resKey.getValue().toString());
  		  String caller = SecurityManager.getManager().getCaller();
  		  if ((caller == null) || (caller.equals("<anonymous>"))) {
  			updateErrorResponse(caXchangeRequestMessage, resKey, CaxchangeErrors.PERMISSION_DENIED_FAULT, "Unable to get the identity of the caller.Caller identity:"+caller);
		  }
          caXchangeRequestMessage.getMetadata().getCredentials().setGridIdentifier(caller);
          try { //Sending the request to the caXchange inbound queue
        	 logger.info("Before sending messge "+new Date().getTime());
             sendMessage(caXchangeRequestMessage);
             logger.info("Before sending messge "+new Date().getTime());
          }catch(Exception e) {
              updateErrorResponse(caXchangeRequestMessage, resKey, CaxchangeErrors.UNKNOWN, "An error occured sending message to the caXchange hub."+e.getMessage());
          }

          if ((responseListeners==null)||(responseListeners.size()==0)) {
              registerResponseListeners();
          }
          logger.debug("Request send to ESB for key "+resKey);

          return ctxResourceHome.getResourceReference(resKey);
      }
      catch(Exception e) {
          logger.error("Error processing message request.",e);
          throw new RemoteException("Error processing message request.",e);
      }
  }
    /**
     * Send the message to the caXchange inbound JMS component.
     *
     * @param caXchangeRequestMessage
     * @throws Exception
     */
    protected void sendMessage(gov.nih.nci.caxchange.Message caXchangeRequestMessage) throws Exception {
        Session session = null;
        try  {
        	CaXchangeRequestProcessorConfiguration configuration = getConfiguration();
            CaXchangeExternalProperties properties = CaXchangeExternalProperties.getInstance();
            String inboundJmsBrokerUrl = properties.getProperty("inbound.jms.brokerURL");
            if (inboundJmsBrokerUrl==null) {
            	inboundJmsBrokerUrl = configuration.getCaXchangeInboundBrokerURL();
            }
            if (connectionFactory == null) {
                connectionFactory = new ActiveMQConnectionFactory(inboundJmsBrokerUrl);
            }
            if (destination == null) {
                destination = new ActiveMQQueue(destinationName);
            }
            if (connection == null) {
               connection = connectionFactory.createConnection();
            }
            try {
               session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }catch(JMSException connectionFailed) {
            	//Connection may have failed try to create a new connection:
            	connection = connectionFactory.createConnection();
            	session  = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }
            MessageProducer producer = session.createProducer(destination);
            RequestGeneratorHelper rgh = new RequestGeneratorHelper();
            rgh.setCaxchangeRequest(caXchangeRequestMessage);
            String req = rgh.getRequestForCaxchange();
            TextMessage message = session.createTextMessage();
            logger.debug("Sending message:"+req);
            message.setText(req);
            
            //commented lines below can be used for simulating synchronous response to the client
            //TemporaryQueue outDest = session.createTemporaryQueue();
            //message.setJMSReplyTo(outDest);
            producer.send(message);
            
            //MessageConsumer consumer = session.createConsumer(outDest);
            //Message msgBack = null;
            //msgBack = consumer.receive(12000);
{'fraction_non_alphanumeric': 0.09200413833875258, 'fraction_numerical': 0.0018474726574046704, 'mean_word_length': 3.5827971554351508, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '1470357', 'n_tokens_mistral': 3634, 'n_tokens_neox': 3590, 'n_words': 709}