Close connection when a framing error is encountered during decoding.

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

2 files changed, 37 insertions, 29 deletions

diff --git a/cpp/src/qpid/sys/AsynchIOAcceptor.cpp b/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
index 0700fff8af..42272a4ac9 100644
--- a/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
+++ b/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
@@ -89,12 +89,14 @@ class AsynchIOHandler : public qpid::sys::ConnectionOutputHandler {
 	Mutex frameQueueLock;
 	bool frameQueueClosed;
 	bool initiated;
-	
+        bool readError;
+
 public:
 	AsynchIOHandler() :
 		inputHandler(0),
 		frameQueueClosed(false),
-		initiated(false)
+		initiated(false),
+                readError(false)
 	{}
 	
 	~AsynchIOHandler() {
@@ -201,36 +203,41 @@ void AsynchIOHandler::close() {
 
 // Input side
 void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) {
-	framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
+    if (readError) {
+        return;
+    }
+    framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
     if(initiated){
         framing::AMQFrame frame;
         try{
             while(frame.decode(in)) {
                 QPID_LOG(debug, "RECV: " << frame);
-		        inputHandler->received(frame);
-    		}
-		}catch(const std::exception& e){
-	    	QPID_LOG(error, e.what());
-		}
-	}else{
-	    framing::ProtocolInitiation protocolInit;
-	    if(protocolInit.decode(in)){
-	        QPID_LOG(debug, "INIT [" << aio << "]");
-	        inputHandler->initiated(protocolInit);
-	        initiated = true;
-	    }
-	}
-	// TODO: unreading needs to go away, and when we can cope
-	// with multiple sub-buffers in the general buffer scheme, it will
-	if (in.available() != 0) {
-		// Adjust buffer for used bytes and then "unread them"
-		buff->dataStart += buff->dataCount-in.available();
-		buff->dataCount = in.available();
-		aio->unread(buff);
-	} else {
-		// Give whole buffer back to aio subsystem
-		aio->queueReadBuffer(buff);
-	}
+                inputHandler->received(frame);
+            }
+        }catch(const std::exception& e){
+            QPID_LOG(error, e.what());
+            readError = true;
+            aio->queueWriteClose();
+        }
+    }else{
+        framing::ProtocolInitiation protocolInit;
+        if(protocolInit.decode(in)){
+            QPID_LOG(debug, "INIT [" << aio << "]");
+            inputHandler->initiated(protocolInit);
+            initiated = true;
+        }
+    }
+    // TODO: unreading needs to go away, and when we can cope
+    // with multiple sub-buffers in the general buffer scheme, it will
+    if (in.available() != 0) {
+        // Adjust buffer for used bytes and then "unread them"
+        buff->dataStart += buff->dataCount-in.available();
+        buff->dataCount = in.available();
+        aio->unread(buff);
+    } else {
+        // Give whole buffer back to aio subsystem
+        aio->queueReadBuffer(buff);
+    }
 }
 
 void AsynchIOHandler::eof(AsynchIO&) {
@@ -294,7 +301,7 @@ void AsynchIOHandler::idle(AsynchIO&){
 	} while (!frameQueue.empty());
 
 	if (frameQueueClosed) {
-		aio->queueWriteClose();
+            aio->queueWriteClose();
 	}
 }
 
diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
index 3512363d46..ffb7c867e4 100644
--- a/cpp/src/qpid/sys/posix/AsynchIO.cpp
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -157,7 +157,8 @@ void AsynchIO::queueWrite(BufferBase* buff) {
 }
 
 void AsynchIO::queueWriteClose() {
-	queuedClose = true;
+    queuedClose = true;
+    DispatchHandle::rewatchWrite();
 }
 
 /** Return a queued buffer if there are enough