Connect to JCR Repostory and JMS queue

@throws JMSException if JMS Exception occurred
1868
@PostConstruct
    public void acquireConnections() throws JMSException {
        LOGGER.debug("Initializing: {}", this.getClass().getCanonicalName());

        connection = connectionFactory.createConnection();
        connection.start();
        jmsSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = jmsSession.createProducer(createDestination());
        eventBus.register(this);
    }
940650
@Override
    public JMXConnector newJMXConnector(JMXServiceURL serviceURL, Map<String, ?> environment) throws IOException {
        if (serviceURL == null || environment == null)
            throw new NullPointerException();
        if (!ClientProvider.REST_PROTOCOL.equals(serviceURL.getProtocol()))
            throw new MalformedURLException();
        return new Connector(serviceURL, environment);
    }
358360
@Override
	public final void setMessageListener(MessageListener messageListener) throws JMSException
    {
        super.setMessageListener(messageListener);
        
        // If the connection was already started, wake up the new listener in case there were messages
        // waiting in the destination
        if (messageListener != null && connection.isStarted())
        	engine.getDeliveryAsyncTaskManager().execute(wakeUpAsyncTask);
    }
94556
@Override
	public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalQueueReceiver receiver = new LocalQueueReceiver(engine,this,queue,messageSelector,idProvider.createID());
	        registerConsumer(receiver);
	        receiver.initDestination();
	        return receiver;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
1004265
@Subscribe
    public void publishJCREvent(final FedoraEvent fedoraEvent) throws JMSException {
        LOGGER.debug("Received an event from the internal bus.");
        final Message tm =
                eventFactory.getMessage(fedoraEvent, jmsSession);
        LOGGER.debug("Transformed the event to a JMS message.");
        producer.send(tm);

        LOGGER.debug("Put event: {} onto JMS.", tm.getJMSMessageID());
    }
790497
private boolean connect() {
        if (this.factory == null) {
            return false;
        }
        
        if (isConnected()) {
            return true;
        }
        
        try {
            this.connection = this.factory.createConnection();
            this.session = (TopicSession) this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            this.connection.start();
            return true;
        } catch (Exception e) {
            log.error("Error communicating with JMS server: " + e.getMessage());
            disconnect();
            return false;
        }
    }
276802
JmsConnectionImpl instantiateConnection(JmsJcaConnection jcaConnection, Map _passThruProps) throws JMSException {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "instantiateConnection", jcaConnection);
    JmsQueueConnectionImpl jmsQueueConnection = new JmsQueueConnectionImpl(jcaConnection, isManaged(), _passThruProps);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "instantiateConnection",  jmsQueueConnection);
    return jmsQueueConnection;
  }
425536
@Override
	public QueueSender createSender(Queue queue) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalQueueSender sender = new LocalQueueSender(this,queue,idProvider.createID());
	        registerProducer(sender);
	        return sender;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
558532
@Override
    public void commit() throws IllegalStateRuntimeException, TransactionRolledBackRuntimeException, JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "commit");

        try {
            jmsSession.commit();
        } catch (IllegalStateException ise) {
            throw (IllegalStateRuntimeException) JmsErrorUtils.getJMS2Exception(ise, IllegalStateRuntimeException.class);
        } catch (TransactionRolledBackException trbe) {
            throw (TransactionRolledBackRuntimeException) JmsErrorUtils.getJMS2Exception(trbe, TransactionRolledBackRuntimeException.class);
        } catch (JMSException jmse) {
            throw (JMSRuntimeException) JmsErrorUtils.getJMS2Exception(jmse, JMSRuntimeException.class);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "commit");
        }

    }
1137150
@Override
    public void setMessageListener(MessageListener listener) throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setMessageListener", new Object[] { listener });

        try {
            messageConsumer.setMessageListener(listener);
        } catch (JMSException jmse) {
            throw (JMSRuntimeException) JmsErrorUtils.getJMS2Exception(jmse, JMSRuntimeException.class);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "setMessageListener");
        }

    }
875422
@Override
	public final AbstractMessage receiveFromDestination(long timeout, boolean duplicateRequired) throws JMSException
    {
    	synchronized (receiveLock)
		{
	        if (closed)
	            return null; // [JMS SPEC]
	        
	        if (receiving)
	        	throw new FFMQException("Consumer should not be accessed by more than one thread","ILLEGAL_USAGE");
	        
	        receiving = true;
	        try
	        {
		        MessageSelector selector = getReceiveSelector();
		        
		        // No-wait simplified case
		        if (timeout == 0)
		        {
		        	if (!connection.isStarted())
		        		return null;
		        	
		        	AbstractMessage message = localQueue.get((LocalSession)session,
						            						 transactionSet,
					                                         selector);        	
		            if (message == null)
		            	return null;
		            
		        	if (traceEnabled)
		                log.trace(session+" [GET] in "+localQueue+" - "+message);
		
		        	if (duplicateRequired)
		        		message = MessageTools.duplicate(message);
		            message.markAsReadOnly();
		            
		            return message;
		        }
			
		        // Wait loop
		        long now = System.currentTimeMillis();
		        long startTime = now;
		        
		        // Passive wait
		        while (!closed && (timeout < 0 || (now - startTime < timeout)))
		        {
		        	// Don't do anything if connection is not started
		    		if (connection.isStarted())
		    		{
		                // Try obtaining something from target queue
		    			AbstractMessage message = localQueue.get((LocalSession)session,
		    					                                 transactionSet,
		    					                                 selector);
			            if (message != null)
			            {
			                if (traceEnabled)
			                    log.trace(session+" [GET] in "+localQueue+" - "+message);
			                
			                if (duplicateRequired)
				        		message = MessageTools.duplicate(message);
				            message.markAsReadOnly();
				            
				            return message;
			            }
		    		}
		            
		            if (traceEnabled)
		                log.trace("Entering passive wait on "+localQueue+" (timeout="+timeout+")");
		            
		            try
		            {
		                if (timeout <= 0)
		                	receiveLock.wait();
		                else
		                	receiveLock.wait(timeout - (now - startTime));
		            }
		            catch (InterruptedException e)
		            {
		                return null;
		            }
		
		            now = System.currentTimeMillis();
		        }
		        
		        return null;
	        }
	        finally
	        {
	        	receiving = false;
	        }
		}
    }
1309889
@Override
    public JmsQueueConnectionFactory createQueueConnectionFactory() throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createQueueConnectionFactory");
        JmsQueueConnectionFactory jmsqcf = null;

        // get a jca managed queue connection factory, which is used to construct
        // the jms queue connection factory.
        JmsJcaManagedQueueConnectionFactory jcamqcf = JmsJcaFactory.getInstance().createManagedQueueConnectionFactory();

        // invoke the construction of a jms queue connection factory - the jca
        // managed queue connection factory will call back into the jms api code
        // by getting a JmsRAFactoryFactory, and calling a create jms queue
        // connection factory method on it with jca connection factory and jca
        // managed queue connection factory objects as parameters (the jms qcf
        // will keep references to the jca cf and jca mqcf).
        try {
            jmsqcf = (JmsQueueConnectionFactory) jcamqcf.createConnectionFactory();
        } catch (ResourceException re) {
            // No FFDC code needed
            // d222942 review - no documentation for when a RE will be thrown, so
            // default exception ok. (A quick look at the RA suggests the exception isn't
            // ever thrown).
            // d238447 FFDC review. Comment above suggests an FFDC ok for this case.
            throw (JMSException) JmsErrorUtils.newThrowable(
                                                            JMSException.class,
                                                            "EXCEPTION_RECEIVED_CWSIA0007",
                                                            new Object[] { re, "JmsFactoryFactoryImpl.createQueueConnectionFactory (#1)" },
                                                            re,
                                                            "JmsFactoryFactoryImpl.createQueueConnectionFactory#1",
                                                            this,
                                                            tc
                            );
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createQueueConnectionFactory", jmsqcf);
        return jmsqcf;
    }
1324111
public static Queue asRef( Queue queue ) throws JMSException
    {
        if (queue == null)
            return null;
            
        if (queue instanceof QueueRef)
            return queue;
        
        return new QueueRef(queue.getQueueName());
    }
524566
@Override
	public void check() throws JMSException
	{
		if (jdniInitialContextFactoryName == null)
			throw new InvalidDescriptorException("Missing peer property : 'jdniInitialContextFactoryName'");
		if (jndiConnectionFactoryName == null)
			throw new InvalidDescriptorException("Missing peer property : 'jndiConnectionFactoryName'");
		if (providerURL == null)
			throw new InvalidDescriptorException("Missing peer property : 'providerURL'");
	}
751249
public void start(){

        try {
            jmsServer.setConfiguration(config);
            jmsServer.setJmsConfiguration(jmsConfig);

            jmsServer.start();

            ConnectionFactory connectionFactory = (ConnectionFactory) jmsServer.lookup("/cf");
            if(connectionFactory == null){
                throw new AsyncException("Failed to start EmbeddedJMS server due to previous errors.");
            }

            consumerConnection = connectionFactory.createConnection();
            receiverSessionPool = new SessionPool("Consumer", consumerConnection);

            producerConnection = connectionFactory.createConnection();
            senderSessionPool = new SessionPool("Producer", producerConnection);
            configureListeners(injector, queueConfigsList);
            started = true;
        } catch (Exception e) {
            throw new AsyncException(e);
        }
    }
1241353
@Override
    public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool,
            int maxMessages) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }
434126
private void autoStartConsumer() throws JMSException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "autoStartConsumer", new Object[] { autoStart, jmsConnection.getState() });

        try {
            //check if autoStart is true and connection is not started.
            //If its already started, it implies the first consumer has already started the connection
            if (autoStart && (jmsConnection.getState() == JmsInternalConstants.STOPPED)) {
                // start the connection
                jmsConnection.start();
            }
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "autoStartConsumer");
        }
    }