Java 类javax.jms.QueueReceiver 实例源码

@Test
public void testGetQueue() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueReceiver receiver = session.createReceiver(queue);

    assertNotNull(receiver.getQueue());
    assertSame(queue, receiver.getQueue());

    receiver.close();

    try {
        receiver.getQueue();
        fail("Cannot read topic on closed receiver");
    } catch (IllegalStateException ise) {}
}

/**
 * Close a JMS {@link MessageConsumer}.
 * @param messageConsumer JMS Message Consumer that needs to be closed.
 * @throws JMSException if an error occurs while closing the consumer.
 */
public void closeConsumer(MessageConsumer messageConsumer) throws JMSException {
    if (messageConsumer != null) {
        if (logger.isDebugEnabled()) {
            logger.debug("Closing a JMS Message Consumer of: " + this.connectionFactoryString);
        }
        if ((JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec)) || (JMSConstants.JMS_SPEC_VERSION_2_0
                .equals(jmsSpec))) {
            messageConsumer.close();
        } else {
            if (JMSConstants.JMSDestinationType.QUEUE.equals(this.destinationType)) {
                if (messageConsumer instanceof QueueReceiver) {
                    ((QueueReceiver) messageConsumer).close();
                }
            } else {
                if (messageConsumer instanceof  TopicSubscriber) {
                    ((TopicSubscriber) messageConsumer).close();
                }
            }
        }
    }
}

@Test
public void testGetQueue() throws Exception {
   Connection consumerConnection = null;

   try {
      consumerConnection = createConnection();

      Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      MessageConsumer queueConsumer = consumerSession.createConsumer(queue1);

      Queue q = ((QueueReceiver) queueConsumer).getQueue();

      ProxyAssertSupport.assertEquals(queue1, q);
   } finally {
      if (consumerConnection != null) {
         consumerConnection.close();
      }
   }
}

@Test
public void testGetQueueOnClosedConsumer() throws Exception {
   Connection consumerConnection = null;

   try {
      consumerConnection = createConnection();

      Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      MessageConsumer queueConsumer = consumerSession.createConsumer(queue1);

      queueConsumer.close();

      try {
         ((QueueReceiver) queueConsumer).getQueue();
         Assert.fail("must throw a JMS IllegalStateException");
      } catch (javax.jms.IllegalStateException e) {
         // OK
      }
   } finally {
      if (consumerConnection != null) {
         consumerConnection.close();
      }
   }
}

/**
 * com.sun.ts.tests.jms.ee.all.queueconn.QueueConnTest line 171
 */
@Test
public void testCreateReceiverWithMessageSelector() throws Exception {
   QueueConnection qc = null;

   try {
      qc = createQueueConnection();
      QueueSession qs = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

      QueueReceiver qreceiver = qs.createReceiver(queue1, "targetMessage = TRUE");

      qc.start();

      TextMessage m = qs.createTextMessage();
      m.setText("one");
      m.setBooleanProperty("targetMessage", false);

      QueueSender qsender = qs.createSender(queue1);

      qsender.send(m);

      m.setText("two");
      m.setBooleanProperty("targetMessage", true);

      qsender.send(m);

      TextMessage rm = (TextMessage) qreceiver.receive(1000);

      ProxyAssertSupport.assertEquals("two", rm.getText());
   } finally {
      if (qc != null) {
         qc.close();
      }
      Thread.sleep(2000);
      removeAllMessages(queue1.getQueueName(), true);
      checkEmpty(queue1);
   }
}

/**
 * Create a queue receiver
 *
 * @param queue The queue
 * @return The queue receiver
 * @throws JMSException Thrown if an error occurs
 */
@Override
public QueueReceiver createReceiver(final Queue queue) throws JMSException {
   lock();
   try {
      QueueSession session = getQueueSessionInternal();

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createReceiver " + session + " queue=" + queue);
      }

      QueueReceiver result = session.createReceiver(queue);
      result = new ActiveMQRAQueueReceiver(result, this);

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createdReceiver " + session + " receiver=" + result);
      }

      addConsumer(result);

      return result;
   } finally {
      unlock();
   }
}

/**
 * Create a queue receiver
 *
 * @param queue           The queue
 * @param messageSelector
 * @return The queue receiver
 * @throws JMSException Thrown if an error occurs
 */
@Override
public QueueReceiver createReceiver(final Queue queue, final String messageSelector) throws JMSException {
   lock();
   try {
      QueueSession session = getQueueSessionInternal();

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createReceiver " + session + " queue=" + queue + " selector=" + messageSelector);
      }

      QueueReceiver result = session.createReceiver(queue, messageSelector);
      result = new ActiveMQRAQueueReceiver(result, this);

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createdReceiver " + session + " receiver=" + result);
      }

      addConsumer(result);

      return result;
   } finally {
      unlock();
   }
}

@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();
    }
   }

@Override
public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();

        RemoteQueueReceiver receiver =  new RemoteQueueReceiver(idProvider.createID(),
                                                                this,
                                                                DestinationTools.asRef(queue),
                                                                messageSelector);
        registerConsumer(receiver);
        receiver.remoteInit();
        return receiver;
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

public void testQueueReceiversAndTopicSubscriber() throws Exception
{
    Queue queue = new AMQAnyDestination("ADDR:my-queue; {create: always}");
    Topic topic = new AMQAnyDestination("ADDR:amq.topic/test");

    QueueSession qSession = ((AMQConnection)_connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    QueueReceiver receiver = qSession.createReceiver(queue);

    TopicSession tSession = ((AMQConnection)_connection).createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    TopicSubscriber sub = tSession.createSubscriber(topic);

    Session ssn = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageProducer prod1 = ssn.createProducer(ssn.createQueue("ADDR:my-queue"));
    prod1.send(ssn.createTextMessage("test1"));

    MessageProducer prod2 = ssn.createProducer(ssn.createTopic("ADDR:amq.topic/test"));
    prod2.send(ssn.createTextMessage("test2"));

    Message msg1 = receiver.receive();
    assertNotNull(msg1);
    assertEquals("test1",((TextMessage)msg1).getText());

    Message msg2 = sub.receive();
    assertNotNull(msg2);
    assertEquals("test2",((TextMessage)msg2).getText());  
}

/**
 * Prepares the QueueReceiver.
 */
public  synchronized void initQueueReceiver() throws JMSException, NamingException {
    //first of all we have to inititiate the QueueSession
    //(without this we can't instantiate a QueueReceiver)
    initQueueSession();

    //get the QueueReceiver from our QueueSession
    QueueReceiver queueReceiver = queueSession.createReceiver(queue);
    logger.debug("Queue receiver created");

    //if set we pass our ExtendedMessageListener to the QueueReceiver as MessageListener
    if (messageListener != null) {
        queueReceiver.setMessageListener(messageListener);
    }

    //start listening to JMS
    queueConnection.start();
    logger.debug("Queue connection started");
}

public void start(){
    new Thread(new Runnable() {

        public void run() {
            try {
                QueueConnectionFactory connectionFactory = (QueueConnectionFactory)context.lookup("factory/QueueConnectionFactory");
                QueueConnection queueConnection = connectionFactory.createQueueConnection();
                QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
                Queue queue = queueSession.createQueue(destinationName);
                QueueReceiver queueReceiver = queueSession.createReceiver(queue);
                queueReceiver.setMessageListener(MyAsyncReader.this);

                queueConnection.start();

                System.out.println("thred terminated");


            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }).start();
}

public Map openThread() throws ListenerException {
    Map threadContext = new HashMap();

    try {
        if (!isSessionsArePooled()) {
            QueueSession session = createSession();
            threadContext.put(THREAD_CONTEXT_SESSION_KEY, session);

            QueueReceiver receiver;
            receiver = getServiceReceiver(session);
            threadContext.put(THREAD_CONTEXT_RECEIVER_KEY, receiver);
        }
        return threadContext;
    } catch (IfsaException e) {
        throw new ListenerException(getLogPrefix()+"exception in openThread()", e);
    }
}

protected synchronized String getProviderSelector() {
    if (providerSelector==null && useSelectorsForProviders()) {
        try {
            providerSelector=""; // set default, also to avoid re-evaluation time and time again for lower ifsa-versions.
            if (messageProtocol.equals(IfsaMessageProtocolEnum.REQUEST_REPLY)) {
                providerSelector=IFSAConstants.QueueReceiver.SELECTOR_RR;
            }
            if (messageProtocol.equals(IfsaMessageProtocolEnum.FIRE_AND_FORGET)) {
                providerSelector=IFSAConstants.QueueReceiver.SELECTOR_FF;
            }
        } catch (Throwable t) {
            log.debug(getLogPrefix()+"exception determining selector, probably lower ifsa version, ignoring");
        }
    }
    return providerSelector;
}

public QueueReceiver createReceiver(Queue queue) throws JMSException {
    lock();
    try {
        QueueSession session = getQueueSession();
        if (trace)
            log.trace("createReceiver " + session + " queue=" + queue);
        QueueReceiver result = session.createReceiver(queue);
        result = new JmsQueueReceiver(result, this);
        if (trace)
            log.trace("createdReceiver " + session + " receiver=" + result);
        addConsumer(result);
        return result;
    } finally {
        unlock();
    }
}

public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException {
    lock();
    try {
        QueueSession session = getQueueSession();
        if (trace)
            log.trace("createReceiver " + session + " queue=" + queue + " selector=" + messageSelector);
        QueueReceiver result = session.createReceiver(queue, messageSelector);
        result = new JmsQueueReceiver(result, this);
        if (trace)
            log.trace("createdReceiver " + session + " receiver=" + result);
        addConsumer(result);
        return result;
    } finally {
        unlock();
    }
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueReceiver receiver = session.createReceiver(queue);

    assertNotNull(receiver.toString());
}

@Parameters({"admin-username", "admin-password", "broker-hostname", "broker-port"})
@Test
public void testRetrieveConsumerList(String username, String password,
                                     String hostname, String port) throws Exception {
    String queueName = "testSpecificQueueRetrieval";


    // Create a durable queue using a JMS client
    InitialContext initialContextForQueue = ClientHelper
            .getInitialContextBuilder(username, password, hostname, port)
            .withQueue(queueName)
            .build();

    QueueConnectionFactory connectionFactory
            = (QueueConnectionFactory) initialContextForQueue.lookup(ClientHelper.CONNECTION_FACTORY);
    QueueConnection connection = connectionFactory.createQueueConnection();
    connection.start();

    QueueSession queueSession = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    Queue queue = queueSession.createQueue(queueName);
    QueueReceiver receiver1 = queueSession.createReceiver(queue);
    QueueReceiver receiver2 = queueSession.createReceiver(queue);

    HttpGet httpGet = new HttpGet(apiBasePath + QueuesApiDelegate.QUEUES_API_PATH
                                          + "/" + queueName + "/consumers");

    CloseableHttpResponse response = client.execute(httpGet);
    String body = EntityUtils.toString(response.getEntity());

    ConsumerMetadata[] consumers = objectMapper.readValue(body, ConsumerMetadata[].class);

    Assert.assertEquals(consumers.length, 2, "Number of consumers returned is incorrect.");

    receiver1.close();
    receiver2.close();
    queueSession.close();
    connection.close();
}

@Parameters({"admin-username", "admin-password", "broker-hostname", "broker-port"})
@Test
public void testSpecificConsumerRetrieval(String username, String password,
                                          String hostname, String port) throws Exception {
    String queueName = "testSpecificConsumerRetrieval";

    // Create a durable queue using a JMS client
    InitialContext initialContextForQueue = ClientHelper
            .getInitialContextBuilder(username, password, hostname, port)
            .withQueue(queueName)
            .build();

    QueueConnectionFactory connectionFactory
            = (QueueConnectionFactory) initialContextForQueue.lookup(ClientHelper.CONNECTION_FACTORY);
    QueueConnection connection = connectionFactory.createQueueConnection();
    connection.start();

    QueueSession queueSession = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    Queue queue = queueSession.createQueue(queueName);
    QueueReceiver receiver = queueSession.createReceiver(queue);

    HttpGet getAllConsumers = new HttpGet(apiBasePath + QueuesApiDelegate.QUEUES_API_PATH
                                          + "/" + queueName + "/consumers");

    CloseableHttpResponse response = client.execute(getAllConsumers);
    Assert.assertEquals(response.getStatusLine().getStatusCode(), HttpStatus.SC_OK);
    String body = EntityUtils.toString(response.getEntity());

    ConsumerMetadata[] consumers = objectMapper.readValue(body, ConsumerMetadata[].class);

    Assert.assertTrue(consumers.length > 0, "Number of consumers returned is incorrect.");

    int id = consumers[0].getId();
    HttpGet getConsumer = new HttpGet(apiBasePath + QueuesApiDelegate.QUEUES_API_PATH + "/"
                                              + queueName + "/consumers/" + id);

    response = client.execute(getConsumer);
    Assert.assertEquals(response.getStatusLine().getStatusCode(), HttpStatus.SC_OK);
    String consumerString = EntityUtils.toString(response.getEntity());
    ConsumerMetadata consumerMetadata = objectMapper.readValue(consumerString, ConsumerMetadata.class);

    Assert.assertEquals(consumerMetadata.getId().intValue(), id, "incorrect message id");

    receiver.close();
    queueSession.close();
    connection.close();
}

@Override
public QueueReceiver createReceiver(
                                     Queue queue,
                                     String messageSelector ) throws JMSException {

    return addConsumer( ((QueueSession) session).createReceiver(queue, messageSelector));
}

@Override
public QueueReceiver createReceiver(
                                     Queue queue,
                                     String messageSelector ) throws JMSException {

    return addConsumer(queueSession.createReceiver(queue, messageSelector));
}

@Override
public QueueReceiver createSubscriber() throws JMSException
{
    QueueReceiver recv = ((QueueSession) session).createReceiver((Queue) topic, messageSelector);
    log.debug("Created non-durable subscriber");
    return recv;
}

public synchronized void load() throws GenericServiceException {
    try {
        InitialContext jndi = JNDIContextFactory.getInitialContext(jndiServer);
        QueueConnectionFactory factory = (QueueConnectionFactory) jndi.lookup(jndiName);

        if (factory != null) {
            con = factory.createQueueConnection(userName, password);
            con.setExceptionListener(this);
            session = con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            queue = (Queue) jndi.lookup(queueName);
            if (queue != null) {
                QueueReceiver receiver = session.createReceiver(queue);

                receiver.setMessageListener(this);
                con.start();
                this.setConnected(true);
                Debug.logInfo("Listening to queue [" + queueName + "]...", module);
            } else {
                throw new GenericServiceException("Queue lookup failed.");
            }
        } else {
            throw new GenericServiceException("Factory (broker) lookup failed.");
        }
    } catch (NamingException ne) {
        throw new GenericServiceException("JNDI lookup problems; listener not running.", ne);
    } catch (JMSException je) {
        throw new GenericServiceException("JMS internal error; listener not running.", je);
    } catch (GeneralException ge) {
        throw new GenericServiceException("Problems with InitialContext; listener not running.", ge);
    }
}

/**
 * Creates a new server connection.
 * 
 * @param asDispatchers the dispatchers to forward the incoming
 *            messages to
 * @param inputQueueName the jndi name of the input queue to connect
 *            to
 * @param outputQueueName the jndi name of the output queue to
 *            connect to
 * @throws NamingException if the jndi name could not be looked up
 * @throws JMSException if the jms communication fails
 */
public ServerConnection(
    List<ASDispatcher> asDispatchers,
    String inputQueueName,
    String outputQueueName) throws NamingException, JMSException
{
    this.asDispatchers = asDispatchers;
    INPUT_QUEUE_NAME = inputQueueName;
    OUTPUT_QUEUE_NAME = outputQueueName;

    context = new InitialContext();
    Object tmp = context.lookup("ConnectionFactory");
    QueueConnectionFactory qcf = (QueueConnectionFactory) tmp;
    connection = qcf.createQueueConnection();
    session =
        connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    connection.setExceptionListener(this);
    connection.start();

    // setthe link back to the dispatchers
    for (ASDispatcher asDispatcher: asDispatchers) {
        asDispatcher.setServerConnection(this);
    }

    // log the startup
    MFCLog logItem =
        new MFCLog("MFCControll", "application startup", "MFCControll", "");
    send(logItem);

    // set the asyncronous listener
    Queue inputQueue = (Queue) context.lookup("queue/" + INPUT_QUEUE_NAME);
    QueueReceiver receiver = session.createReceiver(inputQueue);
    receiver.setMessageListener(this);
    log.info("bound to queue/" + INPUT_QUEUE_NAME);
}

@Test
public void testTempQueueDelete() throws Exception {
   connection.start();
   QueueSession queueSession = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

   TemporaryQueue tempQueue = queueSession.createTemporaryQueue();

   ActiveMQConnection newConn = (ActiveMQConnection) factory.createConnection();
   try {
      QueueSession newQueueSession = newConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
      QueueSender queueSender = newQueueSession.createSender(tempQueue);

      Message msg = queueSession.createMessage();
      queueSender.send(msg);

      try {
         QueueReceiver consumer = newQueueSession.createReceiver(tempQueue);
         fail("should have gotten exception but got consumer: " + consumer);
      } catch (JMSException ex) {
         //correct
      }

      connection.close();

      try {
         Message newMsg = newQueueSession.createMessage();
         queueSender.send(newMsg);
      } catch (JMSException e) {
         //ok
      }

   } finally {
      newConn.close();
   }
}

/**
 * Create a new wrapper
 *
 * @param consumer the queue receiver
 * @param session  the session
 */
public ActiveMQRAQueueReceiver(final QueueReceiver consumer, final ActiveMQRASession session) {
   super(consumer, session);

   if (ActiveMQRAQueueReceiver.trace) {
      ActiveMQRALogger.LOGGER.trace("constructor(" + consumer + ", " + session + ")");
   }
}

/**
 * Get queue
 *
 * @return The queue
 * @throws JMSException Thrown if an error occurs
 */
@Override
public Queue getQueue() throws JMSException {
   if (ActiveMQRAQueueReceiver.trace) {
      ActiveMQRALogger.LOGGER.trace("getQueue()");
   }

   checkState();
   return ((QueueReceiver) consumer).getQueue();
}

public synchronized void load() throws GenericServiceException {
    try {
        InitialContext jndi = JNDIContextFactory.getInitialContext(jndiServer);
        QueueConnectionFactory factory = (QueueConnectionFactory) jndi.lookup(jndiName);

        if (factory != null) {
            con = factory.createQueueConnection(userName, password);
            con.setExceptionListener(this);
            session = con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            queue = (Queue) jndi.lookup(queueName);
            if (queue != null) {
                QueueReceiver receiver = session.createReceiver(queue);

                receiver.setMessageListener(this);
                con.start();
                this.setConnected(true);
                Debug.logInfo("Listening to queue [" + queueName + "]...", module);
            } else {
                throw new GenericServiceException("Queue lookup failed.");
            }
        } else {
            throw new GenericServiceException("Factory (broker) lookup failed.");
        }
    } catch (NamingException ne) {
        throw new GenericServiceException("JNDI lookup problems; listener not running.", ne);
    } catch (JMSException je) {
        throw new GenericServiceException("JMS internal error; listener not running.", je);
    } catch (GeneralException ge) {
        throw new GenericServiceException("Problems with InitialContext; listener not running.", ge);
    }
}

public synchronized void load() throws GenericServiceException {
    try {
        InitialContext jndi = JNDIContextFactory.getInitialContext(jndiServer);
        QueueConnectionFactory factory = (QueueConnectionFactory) jndi.lookup(jndiName);

        if (factory != null) {
            con = factory.createQueueConnection(userName, password);
            con.setExceptionListener(this);
            session = con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            queue = (Queue) jndi.lookup(queueName);
            if (queue != null) {
                QueueReceiver receiver = session.createReceiver(queue);

                receiver.setMessageListener(this);
                con.start();
                this.setConnected(true);
                Debug.logInfo("Listening to queue [" + queueName + "]...", module);
            } else {
                throw new GenericServiceException("Queue lookup failed.");
            }
        } else {
            throw new GenericServiceException("Factory (broker) lookup failed.");
        }
    } catch (NamingException ne) {
        throw new GenericServiceException("JNDI lookup problems; listener not running.", ne);
    } catch (JMSException je) {
        throw new GenericServiceException("JMS internal error; listener not running.", je);
    } catch (GeneralException ge) {
        throw new GenericServiceException("Problems with InitialContext; listener not running.", ge);
    }
}

public void testCreateQueueReceiver() throws JMSException
{
    QueueReceiver receiver = _session.createQueueReceiver(_queue);
    assertEquals("Queue names should match from QueueReceiver", _queue.getQueueName(), receiver.getQueue().getQueueName());

    receiver = _session.createQueueReceiver(_queue, "abc");
    assertEquals("Queue names should match from QueueReceiver with selector", _queue.getQueueName(), receiver.getQueue().getQueueName());
}

public void testCreateReceiver() throws JMSException
{
    QueueReceiver receiver = _session.createReceiver(_queue);
    assertEquals("Queue names should match from QueueReceiver", _queue.getQueueName(), receiver.getQueue().getQueueName());

    receiver = _session.createReceiver(_queue, "abc");
    assertEquals("Queue names should match from QueueReceiver with selector", _queue.getQueueName(), receiver.getQueue().getQueueName());
}

/**
 * @see javax.jms.QueueSession#createReceiver(javax.jms.Queue)
 */
@Override
public QueueReceiver createReceiver(Queue queue) throws JMSException {
    checkClosed();
    checkDestination(queue);
    JmsDestination dest = JmsMessageTransformation.transformDestination(connection, queue);
    JmsQueueReceiver result = new JmsQueueReceiver(getNextConsumerId(), this, dest, null);
    result.init();
    return result;
}

/**
 * @see javax.jms.QueueSession#createReceiver(javax.jms.Queue, java.lang.String)
 */
@Override
public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException {
    checkClosed();
    checkDestination(queue);
    messageSelector = checkSelector(messageSelector);
    JmsDestination dest = JmsMessageTransformation.transformDestination(connection, queue);
    JmsQueueReceiver result = new JmsQueueReceiver(getNextConsumerId(), this, dest, messageSelector);
    result.init();
    return result;
}

public void rebindPipe(Pipe pipe) throws JMSException {
    if (pipe == null) {
        log.warn("Try to rebind a null pointer pipe.");
        return;
    }
    MessageProducer producer = null;
    if (pipe.getDest() instanceof Topic) {
        pipe.setDest(session.createTopic(((Topic) pipe.getDest()).getTopicName()));
        if (pipe.getConsumer() != null) {
            TopicSubscriber ts = null;
            if (pipe.getMessageSelector() != null && !pipe.getMessageSelector().trim().isEmpty())
                ts = ((TopicSession) session).createSubscriber((Topic) pipe.getDest(), pipe.getMessageSelector(), false);
            else
                ts = ((TopicSession) session).createSubscriber((Topic) pipe.getDest());
            pipe.setConsumer(ts);
            pipe.getConsumer().setMessageListener(pipe.getListener());
        }
        producer = session.createProducer(pipe.getDest());
        producer.setTimeToLive(Constants.ONE_MINUTE * 5);
        pipe.setProducer(producer);
    } else if (pipe.getDest() instanceof Queue) {
        String qn = ((Queue) pipe.getDest()).getQueueName();
        pipe.setDest(session.createQueue(qn));
        if (pipe.getConsumer() != null) {
            QueueReceiver qr = null;
            if (pipe.getMessageSelector() != null && !pipe.getMessageSelector().trim().isEmpty())
                qr = ((QueueSession) session).createReceiver((Queue) pipe.getDest(), pipe.getMessageSelector());
            else
                qr = ((QueueSession) session).createReceiver((Queue) pipe.getDest());
            pipe.setConsumer(qr);
            pipe.getConsumer().setMessageListener(pipe.getListener());
        }
        producer = session.createProducer(pipe.getDest());
        producer.setTimeToLive(Constants.ONE_MINUTE * 5);
        pipe.setProducer(producer);
    }
}

/**
 * @param queue
 * @return QueueRecevier
 * @throws JMSException
 * @see javax.jms.QueueSession#createReceiver(javax.jms.Queue)
 */
@Override
public QueueReceiver createReceiver(Queue queue) throws JMSException {
    checkClosed();
    checkDestination(queue);
    JmsDestination dest = JmsMessageTransformation.transformDestination(connection, queue);
    JmsQueueReceiver result = new JmsQueueReceiver(getNextConsumerId(), this, dest, "");
    result.init();
    return result;
}

/**
 * @param queue
 * @param messageSelector
 * @return QueueReceiver
 * @throws JMSException
 * @see javax.jms.QueueSession#createReceiver(javax.jms.Queue,
 *      java.lang.String)
 */
@Override
public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException {
    checkClosed();
    checkDestination(queue);
    messageSelector = checkSelector(messageSelector);
    JmsDestination dest = JmsMessageTransformation.transformDestination(connection, queue);
    JmsQueueReceiver result = new JmsQueueReceiver(getNextConsumerId(), this, dest, messageSelector);
    result.init();
    return result;
}

protected QueueReceiver getReceiver(Map threadContext, QueueSession session) throws ListenerException {
    if (isSessionsArePooled()) {
        try {
            return getServiceReceiver(session);
        } catch (IfsaException e) {
            throw new ListenerException(getLogPrefix()+"exception creating QueueReceiver", e);
        }
    } 
    return (QueueReceiver) threadContext.get(THREAD_CONTEXT_RECEIVER_KEY);
}

protected void releaseReceiver(QueueReceiver receiver) throws ListenerException {
    if (isSessionsArePooled() && receiver != null) {
        try {
            receiver.close();
            // do not write to log, this occurs too often
        } catch (Exception e) {
            throw new ListenerException(getLogPrefix()+"exception closing QueueReceiver", e);
        }
    }
}

public void closeThread(Map threadContext) throws ListenerException {

    if (!isSessionsArePooled()) {
        QueueReceiver receiver = (QueueReceiver) threadContext.remove(THREAD_CONTEXT_RECEIVER_KEY);
        releaseReceiver(receiver);

        QueueSession session = (QueueSession) threadContext.remove(THREAD_CONTEXT_SESSION_KEY);
        closeSession(session);
    }
}

/**
 * Gets the queueReceiver, by utilizing the <code>getInputQueue()</code> method.<br/>
 * For serverside getQueueReceiver() the creating of the QueueReceiver is done
 * without the <code>selector</code> information, as this is not allowed
 * by IFSA.<br/>
 * For a clientconnection, the receiver is done with the <code>getClientReplyQueue</code>
 */
public QueueReceiver getReplyReceiver(QueueSession session, Message sentMessage)
    throws IfsaException {

    if (isProvider()) {
        throw new IfsaException("cannot get ReplyReceiver: Provider cannot act as Requestor");
    } 

    return getMessagingSource().getReplyReceiver(session, sentMessage);
}