r241@34: chirino | 2007-02-23 14:49:05 -0500

Fixed some Stream tests that broke a little with the latest changes..
 Added better network flow control.
 


git-svn-id: https://svn.apache.org/repos/asf/activemq/branches/activemq-4.1@511085 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java

@@ -50,8 +50,6 @@ public class ActiveMQInputStream extends InputStream implements ActiveMQDispatch
     // These are the messages waiting to be delivered to the client
     private final MessageDispatchChannel unconsumedMessages = new MessageDispatchChannel();
 
-    private int deliveredCounter = 0;
-    private MessageDispatch lastDelivered;
     private boolean eosReached;
     private byte buffer[];
     private int pos;
@@ -117,10 +115,6 @@ public class ActiveMQInputStream extends InputStream implements ActiveMQDispatch
     public void close() throws IOException {
         if (!unconsumedMessages.isClosed()) {
             try {
-                if (lastDelivered != null) {
-                    MessageAck ack = new MessageAck(lastDelivered, MessageAck.STANDARD_ACK_TYPE, deliveredCounter);
-                    connection.asyncSendPacket(ack);
-                }
                 dispose();
                 this.connection.syncSendPacket(info.createRemoveCommand());
             } catch (JMSException e) {
@@ -150,16 +144,8 @@ public class ActiveMQInputStream extends InputStream implements ActiveMQDispatch
         if (md == null || unconsumedMessages.isClosed() || md.getMessage().isExpired())
             return null;
 
-        deliveredCounter++;
+        MessageAck ack = new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, 1);
-        if ((0.75 * info.getPrefetchSize()) <= deliveredCounter) {
-            MessageAck ack = new MessageAck(md, MessageAck.STANDARD_ACK_TYPE, deliveredCounter);
         connection.asyncSendPacket(ack);
-            deliveredCounter = 0;
-            lastDelivered = null;
-        } else {
-            lastDelivered = md;
-        }
-
         return (ActiveMQMessage) md.getMessage();
     }
 

activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java

@@ -582,6 +582,9 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
 
     public void dispose() throws JMSException {
         if (!unconsumedMessages.isClosed()) {
+        	
+        	//log.warn("Consumer is being disposed.", new Exception("trace exception."));
+        	
             // Do we have any acks we need to send out before closing?
             // Ack any delivered messages now. (session may still
             // commit/rollback the acks).
@@ -833,9 +836,20 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
     public void dispatch(MessageDispatch md) {
         MessageListener listener = this.messageListener;
         try {
+        	
             synchronized(unconsumedMessages.getMutex()){
-	            if (!unconsumedMessages.isClosed()) {
+	            if (unconsumedMessages.isClosed()) {
-	                if (listener != null && unconsumedMessages.isRunning() ) {
+	            	return;
+            	}
+                if (listener == null || !unconsumedMessages.isRunning() ) {
+                    unconsumedMessages.enqueue(md);
+                    if (availableListener != null) {
+                        availableListener.onMessageAvailable(this);
+                    }
+                    return;
+                }                	
+        	}
+        	
 	        ActiveMQMessage message = createActiveMQMessage(md);
 	        beforeMessageIsConsumed(md);
 	        try {
@@ -850,14 +864,8 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
 	            }
 	            log.warn("Exception while processing message: " + e, e);
 	        }
-	                } else {
+        	
-	                    unconsumedMessages.enqueue(md);
+            
-	                    if (availableListener != null) {
-	                        availableListener.onMessageAvailable(this);
-	                    }
-	                }
-	            }
-            }
         } catch (Exception e) {
         	session.connection.onAsyncException(e);
         }

activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -790,13 +790,10 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
                 MessageDispatch md=(MessageDispatch) command;
                 Runnable sub=(Runnable) md.getConsumer();
                 broker.processDispatch(md);
-                try{
-                    dispatch(command);
-                }finally{
                 if(sub!=null){
                     sub.run();
                 }
-                }
+                dispatch(command);
             } else if( command.isShutdownInfo() ) {
                 dispatch(command);
                 dispatchStopped.countDown();

activemq-core/src/main/java/org/apache/activemq/broker/jmx/ConnectionViewMBean.java

@@ -71,4 +71,11 @@ public interface ConnectionViewMBean extends Service {
      */
     public String getRemoteAddress();
 
+    /**
+     * Returns the JMS connection id for this connection
+     * 
+     * @return the JMS connection id for this connection
+     */
+    public String getConnectionId();
+
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -280,8 +280,8 @@ public class Queue implements Destination {
 
     public void send(final ConnectionContext context, final Message message) throws Exception {
 
-        if (context.isProducerFlowControl() && !context.isNetworkConnection()) {
+        if (context.isProducerFlowControl() ) {
-            if( message.isResponseRequired() ) {
+            if( message.isResponseRequired() || context.isNetworkConnection() ) {
             	if( usageManager.isFull() ) {
 //            		System.out.println("Registering callback...");
 	            	Runnable callback = new Runnable() {

activemq-core/src/main/java/org/apache/activemq/broker/region/TempQueueRegion.java

@@ -38,7 +38,6 @@ public class TempQueueRegion extends AbstractRegion {
         super(broker,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
         // We should allow the following to be configurable via a Destination Policy 
         // setAutoCreateDestinations(false);
-        System.out.println("test");
     }
 
     protected Destination createDestination(ConnectionContext context, ActiveMQDestination destination) throws Exception {

activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java

@@ -232,7 +232,7 @@ public class Topic implements Destination {
 
     public void send(final ConnectionContext context, final Message message) throws Exception {
 
-        if (context.isProducerFlowControl()  && !context.isNetworkConnection() ) {
+        if (context.isProducerFlowControl() ) {
             if (usageManager.isSendFailIfNoSpace() && usageManager.isFull()) {
                 throw new javax.jms.ResourceAllocationException("Usage Manager memory limit reached");
             } else {

activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -490,8 +490,7 @@ public abstract class DemandForwardingBridgeSupport implements Bridge {
                             log.trace("bridging "+localBrokerName+" -> "+remoteBrokerName+": "+message);
                         
                         
-                        
+                        if( !( message.isResponseRequired() || message.getDestination().isQueue() ) ) {
-                        if( !message.isResponseRequired() ) {
                             
                             // If the message was originally sent using async send, we will preserve that QOS
                             // by bridging it using an async send (small chance of message loss).

activemq-core/src/test/java/org/apache/activemq/LargeStreamletTest.java

@@ -17,6 +17,7 @@ package org.apache.activemq;
 * limitations under the License.
 */
 
+import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -59,14 +60,13 @@ public final class LargeStreamletTest extends TestCase {
         final ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
                 BROKER_URL);
 
-        final ActiveMQConnection connection = (ActiveMQConnection) factory
+        final ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection();
-                .createConnection();
         connection.start();
+        final ActiveMQConnection connection2 = (ActiveMQConnection) factory.createConnection();
+        connection2.start();
+        
         try {
-            final Session session = connection.createSession(false,
+            final Destination destination = new ActiveMQQueue("streamtest");
-                    Session.AUTO_ACKNOWLEDGE);
-            try {
-                final Destination destination = session.createQueue("wibble");
             final Thread readerThread = new Thread(new Runnable() {
 
                 public void run() {
@@ -99,7 +99,7 @@ public final class LargeStreamletTest extends TestCase {
                     totalWritten.set(0);
                     int count = MESSAGE_COUNT;
                     try {
-                            final OutputStream outputStream = connection
+                        final OutputStream outputStream = connection2
                                 .createOutputStream(destination);
                         try {
                             final byte[] buf = new byte[BUFFER_SIZE];
@@ -147,11 +147,9 @@ public final class LargeStreamletTest extends TestCase {
             Assert.assertEquals("Not all messages accounted for", 
                     totalWritten.get(), totalRead.get());
                 
-            } finally {
-                session.close();
-            }
         } finally {
             connection.close();
+            connection2.close();
         }
     }
 

activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java

@@ -135,7 +135,7 @@ public class ProducerFlowControlTest extends JmsTestSupport {
     protected BrokerService createBroker() throws Exception {
         BrokerService service = new BrokerService();
         service.setPersistent(false);
-        service.setUseJmx(false);
+        service.setUseJmx(true);
         
         // Setup a destination policy where it takes only 1 message at a time.
         PolicyMap policyMap = new PolicyMap();