nk4um

public class JMSTransport extends NKFTransportImpl implements MessageListener
{
   private String mDescription="JMSTransport";
   private WorkerThreadPool mPool;
   private IURRepresentation mConfig;
   private List mConsumers;
   private List mSessions;
   private HashSet mNeedAck;
    
   //Used if JMS Header is not sent in the message
    private String defaultDestinationName = "DEFAULT.INBOUND";
   
   /** Return a generated description of the transport */
   public String getDescription()
   {
[Unmodified]   
   }
   
   /** Starts the transport
    */
   protected void startTransport() throws Exception
[Unmodified]   
   }
   
   /** Closes connections and stops the transport
    */
   protected void stopTransport() throws Exception
   {   
[Unmodified]   
   }
   
   /** Called once a second to check for updates to configuration
    */
   private void innerProcess() throws Exception
   {   
[Unmodified]
   }
   
   /** Called if the configuration changes.
    * initializes internal data structures and connects to all queues and topics with a listener
    */
   private void initialiseConfig() throws Exception
   {   
      JMSConnectionAspect connection = (JMSConnectionAspect)mConfig.getAspect(JMSConnectionAspect.class);
      cleanup(connection);
      QueueConnection qc = connection.getQueueConnection();
      TopicConnection tc = connection.getTopicConnection();
      
      // initialise new listener
      List names = connection.getNames();
      mConsumers = new ArrayList();
      mSessions = new ArrayList();
      mNeedAck = new HashSet();
      mDescription = "JMSTransport:";
      for (Iterator i=names.iterator(); i.hasNext(); )
      {   String name = (String)i.next();
         try
         {   Destination dest = (Destination) connection.getJNDIContext().lookup(name);
            String physicalName;
            IXDAReadOnly config = connection.getConfiguration(name);
            boolean ackOnException = config.getText("ackOnException",true).toLowerCase().equals("true");
            int acknowedgeMode = ackOnException?Session.AUTO_ACKNOWLEDGE:Session.CLIENT_ACKNOWLEDGE;
            String messageSelector=null;
            try
            {   messageSelector = config.getText("messageSelector",true);
            } catch (Exception e)
            { /* no selector */ }
            Session s;
            
                if (dest instanceof Queue)
            {   if (qc==null) throw new Exception("No QueueConnectionFactory registered");
               s = qc.createQueueSession(false, acknowedgeMode);
               physicalName = ((Queue)dest).getQueueName();
            }
            else if (dest instanceof Topic)
            {   if (tc==null) throw new Exception("No TopicConnectionFactory registered");
               s = tc.createTopicSession(false, acknowedgeMode);
               physicalName = ((Topic)dest).getTopicName();
            }
            else throw new NKFException("Destination must be queue or topic");
            
                if (!ackOnException)
            {   mNeedAck.add(this.retrieveDestinationName(physicalName));
            }
            
                MessageConsumer consumer = s.createConsumer(dest,messageSelector);
            consumer.setMessageListener(this);
            mSessions.add(s);
            mConsumers.add(consumer);
            mDescription+=" "+name;
         }
         catch (Exception e)
         {   SysLogger.log1(SysLogger.WARNING, this, "Failed to initialise consumer for %1",name);
            SysLogger.log(SysLogger.WARNING, this, e.toString());
         }
      }
      if (qc!=null) qc.start();
      if (tc!=null) tc.start();
   }
   
   /** called on stop and when configuration changes to close existing connections and sessions
    */
   private void cleanup(JMSConnectionAspect aConnection) throws Exception
   {
[Unmodified]   
   }
   
   /** Called when a message arrives on a topic or queue we are listening to
    */
   public void onMessage(Message aMessage)
   {   String uriString=null;
      try
      {   
          Destination dest = aMessage.getJMSDestination();
        
         String name;
         if (dest instanceof Queue)
         {   name = ((Queue)dest).getQueueName();
            uriString="jms-queue-transport:"+name;
                name = this.retrieveDestinationName(name);//retrieve logical name
         }
         else if (dest instanceof Topic)
         {   name = ((Topic)dest).getTopicName();
            uriString="jms-topic-transport:"+name;
                name = this.retrieveDestinationName(name);//retrieve logical name
         }
            //no JMS Header
            else if (dest == null)
            {   name = defaultDestinationName; //set default logical name
                uriString="jms-queue-transport:"+name;
                
            }
         else throw new Exception("Unknown destination type: "+dest.getClass().getName());
         
            dispatchRequest(uriString, aMessage);
         
            if (mNeedAck.contains(name))
         {   aMessage.acknowledge();
         }
      }
      catch (Exception e)
      {
         SysLogger.log2(SysLogger.WARNING, this, "Failed to process message from %1: %2",uriString, e.getMessage());
         SysLogger.log(SysLogger.WARNING, this, e.toString());      
      }
   }   
   
   /** Dispatches the given message to the transport manager for execution
    */
   private void dispatchRequest(String aDestName, Message aMessage) throws Exception
   {   
[Unmodified]
   }
    
    /**
     * 
     * Retrieve the logical destination name from the physical destination name
     * @param physicalName
     * @return String
     */
    private String retrieveDestinationName (String physicalName) {
        
        //method variables
        String desinationName = "UNKNOWN";
        
        String[] nameParts = physicalName.split("\\\\/");
        if (nameParts.length > 0)
            desinationName = nameParts[nameParts.length-1];
        
        return desinationName;
    }
}