Return-Path: 
 <users-return-29161-apmail-activemq-users-archive=activemq.apache.org@activemq.apache.org>
X-Original-To: apmail-activemq-users-archive@www.apache.org
Delivered-To: apmail-activemq-users-archive@www.apache.org
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by minotaur.apache.org (Postfix) with SMTP id B13119CC2
	for <apmail-activemq-users-archive@www.apache.org>;
 Tue, 25 Oct 2011 14:18:46 +0000 (UTC)
Received: (qmail 98368 invoked by uid 500); 25 Oct 2011 14:18:46 -0000
Delivered-To: apmail-activemq-users-archive@activemq.apache.org
Received: (qmail 98340 invoked by uid 500); 25 Oct 2011 14:18:46 -0000
Mailing-List: contact users-help@activemq.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:users-help@activemq.apache.org>
List-Unsubscribe: <mailto:users-unsubscribe@activemq.apache.org>
List-Post: <mailto:users@activemq.apache.org>
List-Id: <users.activemq.apache.org>
Reply-To: users@activemq.apache.org
Delivered-To: mailing list users@activemq.apache.org
Received: (qmail 98332 invoked by uid 99); 25 Oct 2011 14:18:46 -0000
Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136)
    by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 25 Oct 2011 14:18:46 +0000
X-ASF-Spam-Status: No, hits=2.3 required=5.0
	tests=SPF_SOFTFAIL,URI_HEX
X-Spam-Check-By: apache.org
Received-SPF: softfail (athena.apache.org: transitioning domain of
 mkurecka@fractech.net does not designate 216.139.236.26 as permitted sender)
Received: from [216.139.236.26] (HELO sam.nabble.com) (216.139.236.26)
    by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 25 Oct 2011 14:18:41 +0000
Received: from joe.nabble.com ([192.168.236.139])
	by sam.nabble.com with esmtp (Exim 4.72)
	(envelope-from <mkurecka@fractech.net>)
	id 1RIhpc-0006rq-Hv
	for users@activemq.apache.org; Tue, 25 Oct 2011 07:18:20 -0700
Date: Tue, 25 Oct 2011 07:18:20 -0700 (PDT)
From: kureckam <mkurecka@fractech.net>
To: users@activemq.apache.org
Message-ID: <1319552300536-3936864.post@n4.nabble.com>
Subject: Messages not being removed from networked queue
MIME-Version: 1.0
Content-Type: text/plain; charset=us-ascii
Content-Transfer-Encoding: 7bit

I have two activemq brokers networked together. The producer broker (has
static network xml tag) shows enqueued and dequeued values matching when
consumer broker consumes the message, but the dequeued value on the consumer
broker shows zero and if I rerun the consumer it receives all the messages
again. Below is all the code I'm using to test this. Why is the consumer
queue not removing the message from the queue?

// Producer activemq.xml
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:amq="http://activemq.apache.org/schema/core"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">
    <bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.base}/conf/credentials.properties</value>
        </property>      
    </bean>
    <broker xmlns="http://activemq.apache.org/schema/core"
brokerName="transmitBroker" dataDirectory="${activemq.base}/data"
destroyApplicationContextOnStop="true">
        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" producerFlowControl="true"
memoryLimit="1mb">
                  <pendingSubscriberPolicy>
                    <vmCursor />
                  </pendingSubscriberPolicy>
                </policyEntry>
                <policyEntry queue=">" producerFlowControl="true"
memoryLimit="1mb">
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy> 
        <managementContext>
            <managementContext jmxDomainName="transmitDomainName"
connectorPort="1098" />
        </managementContext>

        <networkConnectors>
           <networkConnector uri="static:(tcp://xx.xxx.x.xx:61619)"
duplex="true" />
        </networkConnectors>
        <persistenceAdapter>
            <kahaDB directory="${activemq.base}/data/transmit"/>
        </persistenceAdapter>
        <transportConnectors>
           <transportConnector name="openwire" uri="tcp://0.0.0.0:61618"/>
        </transportConnectors>
    </broker>
    <import resource="jetty.xml"/>
</beans>

// MsgSenderTest.java
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MsgSenderTest
{
   public static void main(final String[] args_)   {
      if(args_.length != 4)      {
         System.out.println("Required parameters;IP, port, Test number and
number of messages");
         System.exit(0);
      }

      final ActiveMQConnectionFactory connectionFactory = new
ActiveMQConnectionFactory("tcp://" + args_[0] + ":" + args_[1]);

      System.out.println("Connecting to ActiveMQ:" +
connectionFactory.getBrokerURL());

      Connection connection = null;
      Session startTopicSession = null;
      MessageProducer startProducer = null;

      try      {
         final int numberOfMessages = Integer.parseInt(args_[3]);

         connection = connectionFactory.createConnection();
         connection.start();

         startTopicSession = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);

         final Queue startQueue = startTopicSession.createQueue("Test" +
args_[2]);

         startProducer = startTopicSession.createProducer(startQueue);

         for(int i = 0; i < numberOfMessages; i++)
         {
            System.out.println("Sending message #" + (i + 1));
            startProducer.send(startTopicSession.createMessage());

            try            {               Thread.sleep(500);            }
            catch(final Exception e2)            {}
         }

         final Message message = startTopicSession.createMessage();

         message.setStringProperty("END", "");
         startProducer.send(message);
      }
      catch(final Exception e)      {         e.printStackTrace();      }
      finally
      {
         if(startProducer != null)
         {
            try            {               startProducer.close();           
}
            catch(final JMSException e)            {              
e.printStackTrace();            }
         }

         if(startTopicSession != null)
         {
            try            {               startTopicSession.close();           
}
            catch(final JMSException e)            {              
e.printStackTrace();            }
         }

         if(connection != null)
         {
            try            {               connection.close();            }
            catch(final JMSException e)            {              
e.printStackTrace();            }
         }
      }
   }
}

// Consumer activemq.xml
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:amq="http://activemq.apache.org/schema/core"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">
    <bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.base}/conf/credentials.properties</value>
        </property>      
    </bean>

    <broker xmlns="http://activemq.apache.org/schema/core"
brokerName="receiveBroker" dataDirectory="${activemq.base}/data"
destroyApplicationContextOnStop="true">
        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" producerFlowControl="true"
memoryLimit="1mb">
                  <pendingSubscriberPolicy>
                    <vmCursor />
                  </pendingSubscriberPolicy>
                </policyEntry>
                <policyEntry queue=">" producerFlowControl="true"
memoryLimit="1mb">
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy> 
 
        <managementContext>
            <managementContext jmxDomainName="receiveDomainName"
connectorPort="1099" />
        </managementContext>

         <networkConnectors>
         </networkConnectors>

        <persistenceAdapter>
            <kahaDB directory="${activemq.base}/data/receive"/>
        </persistenceAdapter>

        <transportConnectors>
           <transportConnector name="openwire" uri="tcp://0.0.0.0:61619"/>
        </transportConnectors>
    </broker>
    <import resource="jetty.xml"/>
</beans>

// MsgListenerTest.java
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

public class MsgListenerTest implements MessageListener
{
   private final ActiveMQConnectionFactory connectionFactory;
   private Connection connection;
   private Session session;
   private MessageConsumer consumer;

   private int msgNumber = 1;

   public static void main(final String[] args_)
   {
      if(args_.length != 3){
         System.out.println("Required parameters: IP, port, test number");
         System.exit(0);
      }

      new MsgListenerTest(args_);
   }

   public MsgListenerTest(final String[] args_){
      this.connectionFactory = new ActiveMQConnectionFactory("tcp://" +
args_[0] + ":" + args_[1]);

      final String queueName = "Test" + args_[2];

      try
      {
         this.connection = this.connectionFactory.createConnection();
         this.connection.start();

         this.session = this.connection.createSession(true,
Session.AUTO_ACKNOWLEDGE);

         this.consumer =
this.session.createConsumer(this.session.createQueue(queueName));

         this.consumer.setMessageListener(this);
      }
      catch(final Exception e){e.printStackTrace();}
   }

   @Override
   public void onMessage(final Message message_)
   {
      try
      {
         if(message_.getStringProperty("END") == null)
         {
            System.out.println("Received messaage #" + this.msgNumber);

            this.msgNumber++;
         }
         else
         {
            if(this.consumer != null)
            {
               try{this.consumer.close();}
               catch(final JMSException e){e.printStackTrace();}
            }

            if(this.session != null)
            {
               try{this.session.close();}
               catch(final JMSException e){e.printStackTrace();}
            }

            if(this.connection != null)
            {
               try{this.connection.close();}
               catch(final JMSException e) {e.printStackTrace();}
            }
         }
      }
      catch(final JMSException e2){e2.printStackTrace();}
   }
}

--
View this message in context: http://activemq.2283324.n4.nabble.com/Messages-not-being-removed-from-networked-queue-tp3936864p3936864.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.