* Added support for channel.flow:

    cpp/tests/ChannelTest.cpp
    cpp/lib/broker/SessionHandlerImpl.cpp
    cpp/lib/broker/BrokerChannel.h
    cpp/lib/broker/BrokerChannel.cpp

* Fixed client connection closing process:
    cpp/lib/common/sys/apr/Socket.cpp
    cpp/lib/client/Connector.h
    cpp/lib/client/Connector.cpp
    cpp/lib/client/Connection.h
    cpp/lib/client/Connection.cpp

* Use amq.direct rather than default exchange in P2P test 
  (to interop with java)
    cpp/tests/BasicP2Ptest.h



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2@536584 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 29 insertions, 13 deletions

diff --git a/cpp/lib/client/Connection.cpp b/cpp/lib/client/Connection.cpp
index ad8aa1d0dd..f7897aa4df 100644
--- a/cpp/lib/client/Connection.cpp
+++ b/cpp/lib/client/Connection.cpp
@@ -30,9 +30,11 @@ using namespace qpid::framing;
 using namespace qpid::sys;
 using namespace qpid::sys;
 
-u_int16_t Connection::channelIdCounter;
-
-Connection::Connection(  bool debug, u_int32_t _max_frame_size, qpid::framing::ProtocolVersion* _version) : max_frame_size(_max_frame_size), closed(true),
+Connection::Connection(  bool _debug, u_int32_t _max_frame_size, qpid::framing::ProtocolVersion* _version) : 
+    debug(_debug),
+    channelIdCounter(0),
+    max_frame_size(_max_frame_size), 
+    closed(true),
     version(_version->getMajor(),_version->getMinor())
 {
     connector = new Connector(version, debug, _max_frame_size);
@@ -96,7 +98,7 @@ void Connection::open(const std::string& _host, int _port, const std::string& ui
     }else{
         THROW_QPID_ERROR(PROTOCOL_ERROR, "Bad response");
     }
-    
+    closed = false;
 }
 
 void Connection::close(){
@@ -108,6 +110,7 @@ void Connection::close(){
         
         sendAndReceive(new AMQFrame(version, 0, new ConnectionCloseBody(version, code, text, classId, methodId)), method_bodies.connection_close_ok);
         connector->close();
+        closed = true;
     }
 }
 
diff --git a/cpp/lib/client/Connection.h b/cpp/lib/client/Connection.h
index 05d139e99c..bbf8c03b0b 100644
--- a/cpp/lib/client/Connection.h
+++ b/cpp/lib/client/Connection.h
@@ -68,7 +68,8 @@ namespace client {
 
         typedef std::map<int, Channel*>::iterator iterator;
 
-	static u_int16_t channelIdCounter;
+        const bool debug;
+	u_int16_t channelIdCounter;
 
 	std::string host;
 	int port;
diff --git a/cpp/lib/client/Connector.cpp b/cpp/lib/client/Connector.cpp
index b34e66fd94..c57b3d6dc4 100644
--- a/cpp/lib/client/Connector.cpp
+++ b/cpp/lib/client/Connector.cpp
@@ -57,9 +57,10 @@ void Connector::init(ProtocolInitiation* header){
 }
 
 void Connector::close(){
-    closed = true;
-    socket.close();
-    receiver.join();
+    if (markClosed()) {
+        socket.close();
+        receiver.join();
+    }
 }
 
 void Connector::setInputHandler(InputHandler* handler){
@@ -101,14 +102,24 @@ void Connector::writeToSocket(char* data, size_t available){
 }
 
 void Connector::handleClosed(){
-    closed = true;
-    socket.close();
-    if(shutdownHandler) shutdownHandler->shutdown();
+    if (markClosed()) {
+        socket.close();
+        if(shutdownHandler) shutdownHandler->shutdown();
+    }
+}
+
+bool Connector::markClosed(){
+    if (closed) {
+        return false;
+    } else {
+        closed = true;
+        return true;
+    }
 }
 
 void Connector::checkIdle(ssize_t status){
     if(timeoutHandler){
-         Time t = now() * TIME_MSEC;
+        Time t = now() * TIME_MSEC;
         if(status == Socket::SOCKET_TIMEOUT) {
             if(idleIn && (t - lastIn > idleIn)){
                 timeoutHandler->idleIn();
diff --git a/cpp/lib/client/Connector.h b/cpp/lib/client/Connector.h
index f9e50f3216..eccb931e6c 100644
--- a/cpp/lib/client/Connector.h
+++ b/cpp/lib/client/Connector.h
@@ -44,7 +44,7 @@ namespace client {
 	const int send_buffer_size;
 	qpid::framing::ProtocolVersion version;
 
-	bool closed;
+	volatile bool closed;
 
         int64_t lastIn;
         int64_t lastOut;
@@ -73,6 +73,7 @@ namespace client {
 
 	void run();
 	void handleClosed();
+        bool markClosed();
 
     public:
 	Connector(const qpid::framing::ProtocolVersion& pVersion, bool debug = false, u_int32_t buffer_size = 1024);