content: /**
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.activemq;
  
  import java.util.Enumeration;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  import jakarta.jms.*;
  import jakarta.jms.IllegalStateException;
  
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ConsumerId;
  import org.apache.activemq.command.MessageDispatch;
  import org.apache.activemq.selector.SelectorParser;
  
  /**
   * A client uses a <CODE>QueueBrowser</CODE> object to look at messages on a
   * queue without removing them. <p/>
   * <P>
   * The <CODE>getEnumeration</CODE> method returns a <CODE>
   * java.util.Enumeration</CODE>
   * that is used to scan the queue's messages. It may be an enumeration of the
   * entire content of a queue, or it may contain only the messages matching a
   * message selector. <p/>
   * <P>
   * Messages may be arriving and expiring while the scan is done. The JMS API
   * does not require the content of an enumeration to be a static snapshot of
   * queue content. Whether these changes are visible or not depends on the JMS
   * provider. <p/>
   * <P>
   * A <CODE>QueueBrowser</CODE> can be created from either a <CODE>Session
   * </CODE>
   * or a <CODE>QueueSession</CODE>.
   *
   * @see jakarta.jms.Session#createBrowser
   * @see jakarta.jms.QueueSession#createBrowser
   * @see jakarta.jms.QueueBrowser
   * @see jakarta.jms.QueueReceiver
   */
  
  public class ActiveMQQueueBrowser implements QueueBrowser, Enumeration {
  
      private final ActiveMQSession session;
      private final ActiveMQDestination destination;
      private final String selector;
  
      private ActiveMQMessageConsumer consumer;
      private boolean closed;
      private final ConsumerId consumerId;
      private final AtomicBoolean browseDone = new AtomicBoolean(true);
      private final boolean dispatchAsync;
      private Object semaphore = new Object();
  
      /**
       * Constructor for an ActiveMQQueueBrowser - used internally
       * @throws JMSException
       */
      protected ActiveMQQueueBrowser(ActiveMQSession session, ConsumerId consumerId, ActiveMQDestination destination, String selector, boolean dispatchAsync) throws JMSException {
          if (destination == null) {
              throw new InvalidDestinationException("Don't understand null destinations");
          } else if (destination.getPhysicalName() == null) {
              throw new InvalidDestinationException("The destination object was not given a physical name.");
          }
          if (selector != null && selector.trim().length() != 0) {
              // Validate the selector
              SelectorParser.parse(selector);
          }
  
          this.session = session;
          this.consumerId = consumerId;
          this.destination = destination;
          this.selector = selector;
          this.dispatchAsync = dispatchAsync;
      }
  
      /**
       * @throws JMSException
       */
      private ActiveMQMessageConsumer createConsumer() throws JMSException {
          browseDone.set(false);
          ActiveMQPrefetchPolicy prefetchPolicy = session.connection.getPrefetchPolicy();
  
          return new ActiveMQMessageConsumer(session, consumerId, destination, null, selector, prefetchPolicy.getQueueBrowserPrefetch(), prefetchPolicy
              .getMaximumPendingMessageLimit(), false, true, dispatchAsync, null) {
              public void dispatch(MessageDispatch md) {
                  if (md.getMessage() == null) {
                      browseDone.set(true);
                  } else {
                      super.dispatch(md);
                  }
                  notifyMessageAvailable();
              }
          };
      }
  
      private void destroyConsumer() {
          if (consumer == null) {
              return;
          }
          try {
              if (session.getTransacted() && session.getTransactionContext().isInLocalTransaction()) {
                  session.commit();
              }
              consumer.close();
              consumer = null;
          } catch (JMSException e) {
              e.printStackTrace();
          }
      }
  
      /**
       * Gets an enumeration for browsing the current queue messages in the order
       * they would be received.
       *
       * @return an enumeration for browsing the messages
       * @throws JMSException if the JMS provider fails to get the enumeration for
       *                 this browser due to some internal error.
       */
  
      public Enumeration getEnumeration() throws JMSException {
          checkClosed();
          if (consumer == null) {
              consumer = createConsumer();
          }
          return this;
      }
  
      private void checkClosed() throws IllegalStateException {
          if (closed) {
              throw new IllegalStateException("The Consumer is closed");
          }
      }
  
      /**
       * @return true if more messages to process
       */
      public boolean hasMoreElements() {
          while (true) {
  
              synchronized (this) {
                  if (consumer == null) {
                      return false;
                  }
              }
  
              if (consumer.getMessageSize() > 0) {
                  return true;
              }
  
              if (browseDone.get() || !session.isRunning()) {
                  destroyConsumer();
                  return false;
              }
  
              waitForMessage();
          }
      }
  
      /**
       * @return the next message
       */
      public Object nextElement() {
          while (true) {
  
              synchronized (this) {
                  if (consumer == null) {
                      return null;
                  }
              }
  
              try {
                  jakarta.jms.Message answer = consumer.receiveNoWait();
                  if (answer != null) {
                      return answer;
                  }
              } catch (JMSException e) {
                  this.session.connection.onClientInternalException(e);
                  return null;
              }
  
              if (browseDone.get() || !session.isRunning()) {
                  destroyConsumer();
                  return null;
              }
  
              waitForMessage();
          }
      }
  
      public synchronized void close() throws JMSException {
          browseDone.set(true);
          destroyConsumer();
          closed = true;
      }
  
      /**
       * Gets the queue associated with this queue browser.
       *
       * @return the queue
       * @throws JMSException if the JMS provider fails to get the queue
       *                 associated with this browser due to some internal error.
       */
  
      public Queue getQueue() throws JMSException {
          return (Queue)destination;
      }
  
      public String getMessageSelector() throws JMSException {
          return selector;
      }
  
      // Implementation methods
      // -------------------------------------------------------------------------
  
      /**
       * Wait on a semaphore for a fixed amount of time for a message to come in.
       * @throws JMSException
       */
      protected void waitForMessage() {
          try {
              consumer.sendPullCommand(-1);
              synchronized (semaphore) {
                  semaphore.wait(2000);
              }
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          } catch (JMSException e) {
          }
  
      }
  
      protected void notifyMessageAvailable() {
          synchronized (semaphore) {
              semaphore.notifyAll();
          }
      }
  
      public String toString() {
          return "ActiveMQQueueBrowser { value=" + consumerId + " }";
      }
  
  }
  