QPID-2015: Remove AMQIoTransportProtocolSession. Release the executor service in the same class as it's acquired

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-network-refactor@815729 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 2 insertions, 150 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index 4f396015a2..b776c6ae82 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -706,7 +706,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
             }
             
             _closed = true;
-
+            _poolReference.releaseExecutorService();
             CurrentActor.get().message(_logSubject, ConnectionMessages.CON_1002());
         }
     }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java
deleted file mode 100644
index 8782e00a12..0000000000
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQIoTransportProtocolSession.java
+++ /dev/null
@@ -1,146 +0,0 @@
-package org.apache.qpid.client.protocol;
-/*
- * 
- * 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.
- * 
- */
-
-
-import java.util.UUID;
-
-import javax.security.sasl.SaslClient;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.mina.common.IdleStatus;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.ConnectionTuneParameters;
-import org.apache.qpid.client.handler.ClientMethodDispatcherImpl;
-import org.apache.qpid.client.state.AMQState;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersion;
-import org.apache.qpid.transport.Sender;
-
-public class AMQIoTransportProtocolSession extends AMQProtocolSession
-{
-
-    protected Sender<java.nio.ByteBuffer> _ioSender;
-    private SaslClient _saslClient;
-    private ConnectionTuneParameters _connectionTuneParameters;
-    
-    public AMQIoTransportProtocolSession(AMQProtocolHandler protocolHandler, AMQConnection connection)
-    {
-        super(protocolHandler, connection);
-    }
-    
-    @Override
-    public void closeProtocolSession(boolean waitLast) 
-    {
-        _ioSender.close();
-        _protocolHandler.getStateManager().changeState(AMQState.CONNECTION_CLOSED);
-    }
-
-    @Override
-    public void init()
-    {
-        _ioSender.send(new ProtocolInitiation(_connection.getProtocolVersion()).toNioByteBuffer());
-        _ioSender.flush();
-    }
-
-    @Override
-    protected AMQShortString generateQueueName()
-    {
-        int id;
-        synchronized (_queueIdLock)
-        {
-            id = _queueId++;
-        }
-        return new AMQShortString("tmp_" + UUID.randomUUID() + "_" + id);
-    }
-    
-    @Override
-    public AMQConnection getAMQConnection()
-    {
-        return _connection;
-    }
-    
-    @Override
-    public SaslClient getSaslClient()
-    {
-        return _saslClient;
-    }
-    
-    @Override
-    public void setSaslClient(SaslClient client)
-    {
-        _saslClient = client;
-    }
-    
-    /** @param delay delay in seconds (not ms) */
-    @Override
-    void initHeartbeats(int delay)
-    {
-        if (delay > 0)
-        {
-            // FIXME: actually do something here
-            HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay));
-        }
-    }
-    
-    @Override
-    public void methodFrameReceived(final int channel, final AMQMethodBody amqMethodBody) throws AMQException
-    {
-        // FIXME?
-        _protocolHandler.methodBodyReceived(channel, amqMethodBody, null);
-    }
-    
-    @Override
-    public void writeFrame(AMQDataBlock frame, boolean wait)
-    {      
-        _ioSender.send(frame.toNioByteBuffer());
-        if (wait)
-        {
-            _ioSender.flush();
-        }
-    }
-    
-    @Override
-    public void setSender(Sender<java.nio.ByteBuffer> sender)
-    {
-        _ioSender = sender;
-    }
- 
-    @Override
-    public ConnectionTuneParameters getConnectionTuneParameters()
-    {
-        return _connectionTuneParameters;
-    }
-    
-    @Override
-    public void setConnectionTuneParameters(ConnectionTuneParameters params)
-    {
-        _connectionTuneParameters = params;
-        AMQConnection con = getAMQConnection();
-        con.setMaximumChannelCount(params.getChannelMax());
-        con.setMaximumFrameSize(params.getFrameMax());
-        initHeartbeats((int) params.getHeartbeat());
-    }
-}
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
index c7e2493025..99366101d1 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
@@ -716,6 +716,7 @@ public class AMQProtocolHandler implements ProtocolEngine
                 _logger.debug("FailoverException interrupted connection close, ignoring as connection   close anyway.");
             }
         }
+        _poolReference.releaseExecutorService();
     }
 
     /** @return the number of bytes read from this protocol session */
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java b/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
index 4863611c42..4e02ac3a55 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/pool/PoolingFilter.java
@@ -136,9 +136,6 @@ public abstract class PoolingFilter extends IoFilterAdapter implements Job.JobCo
     public void destroy()
     {
         _logger.debug("Destroy called on PoolingFilter " + toString());
-
-        // When the reference count gets to zero we release the executor service.
-        _poolReference.releaseExecutorService();
     }
 
     /**