Extend credit managers to be aware of transport backpressue

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1644704 13f79535-47bb-0310-9956-ffa450edef68

9 files changed, 44 insertions, 512 deletions

diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java
deleted file mode 100644
index be3a13d2d3..0000000000
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/BytesOnlyCreditManager.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
-*
-* 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.qpid.server.flow;
-
-
-import java.util.concurrent.atomic.AtomicLong;
-
-public class BytesOnlyCreditManager extends AbstractFlowCreditManager
-{
-    private final AtomicLong _bytesCredit;
-
-    public BytesOnlyCreditManager(long initialCredit)
-    {
-        _bytesCredit = new AtomicLong(initialCredit);
-    }
-
-    public long getMessageCredit()
-    {
-        return -1L;
-    }
-
-    public long getBytesCredit()
-    {
-        return _bytesCredit.get();
-    }
-
-    public void restoreCredit(long messageCredit, long bytesCredit)
-    {
-        _bytesCredit.addAndGet(bytesCredit);
-        setSuspended(false);
-    }
-
-    public void removeAllCredit()
-    {
-        _bytesCredit.set(0L);
-    }
-
-    public boolean hasCredit()
-    {
-        return _bytesCredit.get() > 0L;
-    }
-
-    public boolean useCreditForMessage(long msgSize)
-    {
-        if(hasCredit())
-        {
-            if(_bytesCredit.addAndGet(-msgSize) >= 0)
-            {
-                return true;
-            }
-            else
-            {
-                _bytesCredit.addAndGet(msgSize);
-                setSuspended(true);
-                return false;
-            }
-        }
-        else
-        {
-            return false;
-        }
-
-    }
-
-    public void setBytesCredit(long bytesCredit)
-    {
-        _bytesCredit.set( bytesCredit );
-    }
-}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
index 280f2851a4..08aac0b511 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/FlowCreditManager.java
@@ -24,10 +24,6 @@ package org.apache.qpid.server.flow;
 
 public interface FlowCreditManager
 {
-    long getMessageCredit();
-
-    long getBytesCredit();
-
     public static interface FlowCreditManagerListener
     {
         void creditStateChanged(boolean hasCredit);
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java
deleted file mode 100644
index 89fc60666b..0000000000
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/LimitlessCreditManager.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-*
-* 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.qpid.server.flow;
-
-
-public class LimitlessCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
-{
-    public long getMessageCredit()
-    {
-        return -1L;
-    }
-
-    public long getBytesCredit()
-    {
-        return -1L;
-    }
-
-    public void restoreCredit(long messageCredit, long bytesCredit)
-    {
-    }
-
-    public void removeAllCredit()
-    {
-    }
-
-    public boolean hasCredit()
-    {
-        return true;
-    }
-
-    public boolean useCreditForMessage(long msgSize)
-    {
-        return true;
-    }
-}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java
deleted file mode 100644
index 31c1fda968..0000000000
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageAndBytesCreditManager.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
-*
-* 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.qpid.server.flow;
-
-
-public class MessageAndBytesCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
-{
-    private long _messageCredit;
-    private long _bytesCredit;
-
-    public MessageAndBytesCreditManager(final long messageCredit, final long bytesCredit)
-    {
-        _messageCredit = messageCredit;
-        _bytesCredit = bytesCredit;
-    }
-
-    public synchronized long getMessageCredit()
-    {
-        return _messageCredit;
-    }
-
-    public synchronized long getBytesCredit()
-    {
-        return _bytesCredit;
-    }
-
-    public synchronized void restoreCredit(long messageCredit, long bytesCredit)
-    {        
-        _messageCredit += messageCredit;
-        _bytesCredit += bytesCredit;
-        setSuspended(hasCredit());
-    }
-
-    public synchronized void removeAllCredit()
-    {
-        _messageCredit = 0L;
-        _bytesCredit = 0L;
-        setSuspended(true);
-    }
-
-    public synchronized boolean hasCredit()
-    {
-        return (_messageCredit > 0L) && ( _bytesCredit > 0L );
-    }
-
-    public synchronized boolean useCreditForMessage(final long msgSize)
-    {
-        if(_messageCredit == 0L)
-        {
-            setSuspended(true);
-            return false;
-        }
-        else
-        {
-            if(msgSize > _bytesCredit)
-            {
-                setSuspended(true);
-                return false;
-            }
-            _messageCredit--;
-            _bytesCredit -= msgSize;
-            setSuspended(false);
-            return true;
-        }
-        
-    }
-
-    public synchronized void setBytesCredit(long bytesCredit)
-    {
-        _bytesCredit = bytesCredit;
-    }
-}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java
deleted file mode 100644
index 1817e8ad31..0000000000
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/MessageOnlyCreditManager.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
-*
-* 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.qpid.server.flow;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-public class MessageOnlyCreditManager extends AbstractFlowCreditManager implements FlowCreditManager
-{
-    private final AtomicLong _messageCredit;
-
-    public MessageOnlyCreditManager(final long initialCredit)
-    {
-        _messageCredit = new AtomicLong(initialCredit);
-    }
-
-    public long getMessageCredit()
-    {
-        return _messageCredit.get();
-    }
-
-    public long getBytesCredit()
-    {
-        return -1L;
-    }
-
-    public void restoreCredit(long messageCredit, long bytesCredit)
-    {
-        _messageCredit.addAndGet(messageCredit);
-        setSuspended(false);
-
-    }
-
-    public void removeAllCredit()
-    {
-        setSuspended(true);
-        _messageCredit.set(0L);
-    }
-
-    public boolean hasCredit()
-    {
-        return _messageCredit.get() > 0L;
-    }
-
-    public boolean useCreditForMessage(long msgSize)
-    {
-        if(hasCredit())
-        {
-            if(_messageCredit.addAndGet(-1L) >= 0)
-            {
-                setSuspended(false);
-                return true;
-            }
-            else
-            {
-                _messageCredit.addAndGet(1L);
-                setSuspended(true);
-                return false;
-            }
-        }
-        else
-        {
-            setSuspended(true);
-            return false;
-        }
-                
-    }
-
-}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java
deleted file mode 100644
index fc2d4bfb53..0000000000
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/flow/Pre0_10CreditManager.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
-*
-* 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.qpid.server.flow;
-
-
-public class Pre0_10CreditManager extends AbstractFlowCreditManager implements FlowCreditManager
-{
-
-    private volatile long _bytesCreditLimit;
-    private volatile long _messageCreditLimit;
-
-    private volatile long _bytesCredit;
-    private volatile long _messageCredit;
-
-    public Pre0_10CreditManager(long bytesCreditLimit, long messageCreditLimit)
-    {
-        _bytesCreditLimit = bytesCreditLimit;
-        _messageCreditLimit = messageCreditLimit;
-        _bytesCredit = bytesCreditLimit;
-        _messageCredit = messageCreditLimit;
-    }
-
-
-    public synchronized void setCreditLimits(final long bytesCreditLimit, final long messageCreditLimit)
-    {
-        long bytesCreditChange = bytesCreditLimit - _bytesCreditLimit;
-        long messageCreditChange = messageCreditLimit - _messageCreditLimit;
-
-
-
-        if(bytesCreditChange != 0L)
-        {
-            if(bytesCreditLimit == 0L)
-            {
-                _bytesCredit = 0;
-            }
-            else
-            {
-                _bytesCredit += bytesCreditChange;
-            }
-        }
-
-
-        if(messageCreditChange != 0L)
-        {
-            if(messageCreditLimit == 0L)
-            {
-                _messageCredit = 0;
-            }
-            else
-            {
-                _messageCredit += messageCreditChange;
-            }
-        }
-
-
-        _bytesCreditLimit = bytesCreditLimit;
-        _messageCreditLimit = messageCreditLimit;
-
-        setSuspended(!hasCredit());
-
-    }
-
-
-    public long getMessageCredit()
-    {
-        return _messageCredit;
-    }
-
-    public long getBytesCredit()
-    {
-        return _bytesCredit;
-    }
-
-    public synchronized void restoreCredit(final long messageCredit, final long bytesCredit)
-    {
-        final long messageCreditLimit = _messageCreditLimit;
-        boolean notifyIncrease = true;
-        if(messageCreditLimit != 0L)
-        {
-            notifyIncrease = (_messageCredit != 0);
-            long newCredit = _messageCredit + messageCredit;
-            _messageCredit = newCredit > messageCreditLimit ? messageCreditLimit : newCredit;
-        }
-
-
-        final long bytesCreditLimit = _bytesCreditLimit;
-        if(bytesCreditLimit != 0L)
-        {
-            long newCredit = _bytesCredit + bytesCredit;
-            _bytesCredit = newCredit > bytesCreditLimit ? bytesCreditLimit : newCredit;
-            if(notifyIncrease && bytesCredit>0)
-            {
-                notifyIncreaseBytesCredit();
-            }
-        }
-
-
-
-        setSuspended(!hasCredit());
-
-    }
-
-    public synchronized void removeAllCredit()
-    {
-        _bytesCredit = 0L;
-        _messageCredit = 0L;
-        setSuspended(!hasCredit());
-    }
-
-    public synchronized boolean hasCredit()
-    {
-        return (_bytesCreditLimit == 0L || _bytesCredit > 0)
-                && (_messageCreditLimit == 0L || _messageCredit > 0);
-    }
-
-    public synchronized boolean useCreditForMessage(final long msgSize)
-    {
-        if(_messageCreditLimit != 0L)
-        {
-            if(_messageCredit != 0L)
-            {
-                if(_bytesCreditLimit == 0L)
-                {
-                    _messageCredit--;
-
-                    return true;
-                }
-                else
-                {
-                    if((_bytesCredit >= msgSize) || (_bytesCredit == _bytesCreditLimit))
-                    {
-                        _messageCredit--;
-                        _bytesCredit -= msgSize;
-
-                        return true;
-                    }
-                    else
-                    {
-                        return false;
-                    }
-                }
-            }
-            else
-            {
-                setSuspended(true);
-                return false;
-            }
-        }
-        else
-        {
-            if(_bytesCreditLimit == 0L)
-            {
-
-                return true;
-            }
-            else
-            {
-                if((_bytesCredit >= msgSize) || (_bytesCredit == _bytesCreditLimit))
-                {
-                    _bytesCredit -= msgSize;
-
-                    return true;
-                }
-                else
-                {
-                    return false;
-                }
-            }
-
-        }
-
-    }
-}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
index f13af479ad..40aa1bbafd 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
@@ -113,4 +113,6 @@ public interface AMQSessionModel<T extends AMQSessionModel<T,C>, C extends AMQCo
      * @return the time of the last activity or 0 if not in a transaction
      */
     long getTransactionUpdateTime();
+
+    void transportStateChanged();
 }
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
index 6ea9f3600c..3c25e0934c 100755
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngine.java
@@ -157,6 +157,18 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         return _delegate.getSubject();
     }
 
+    @Override
+    public boolean isTransportBlockedForWriting()
+    {
+        return _delegate.isTransportBlockedForWriting();
+    }
+
+    @Override
+    public void setTransportBlockedForWriting(final boolean blocked)
+    {
+        _delegate.setTransportBlockedForWriting(blocked);
+    }
+
     private static final int MINIMUM_REQUIRED_HEADER_BYTES = 8;
 
     public void setNetworkConnection(NetworkConnection network, Sender<ByteBuffer> sender)
@@ -268,6 +280,17 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
         {
             return new Subject();
         }
+
+        @Override
+        public boolean isTransportBlockedForWriting()
+        {
+            return false;
+        }
+
+        @Override
+        public void setTransportBlockedForWriting(final boolean blocked)
+        {
+        }
     }
 
     private class SelfDelegateProtocolEngine implements ServerProtocolEngine
@@ -408,6 +431,17 @@ public class MultiVersionProtocolEngine implements ServerProtocolEngine
             return _delegate.getSubject();
         }
 
+        @Override
+        public boolean isTransportBlockedForWriting()
+        {
+            return false;
+        }
+
+        @Override
+        public void setTransportBlockedForWriting(final boolean blocked)
+        {
+        }
+
         public void exception(Throwable t)
         {
             _logger.error("Error establishing session", t);
diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
index 1c42d9b6fe..47ed224133 100644
--- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
+++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
@@ -456,6 +456,12 @@ public class MockConsumer implements ConsumerTarget
         {
             return 0;
         }
+
+        @Override
+        public void transportStateChanged()
+        {
+
+        }
     }
 
     private static class MockConnectionModel implements AMQConnectionModel
@@ -663,5 +669,7 @@ public class MockConsumer implements ConsumerTarget
         {
 
         }
+
+
     }
 }