qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject svn commit: r584143 - in /incubator/qpid/trunk/qpid/cpp/src/qpid/sys: AsynchIOAcceptor.cpp posix/AsynchIO.cpp
Date Fri, 12 Oct 2007 12:06:39 GMT
Author: gsim
Date: Fri Oct 12 05:06:39 2007
New Revision: 584143

URL: http://svn.apache.org/viewvc?rev=584143&view=rev
Log:
Close connection when a framing error is encountered during decoding.


Modified:
    incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
    incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp?rev=584143&r1=584142&r2=584143&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp Fri Oct 12 05:06:39 2007
@@ -89,12 +89,14 @@
 	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 @@
 
 // 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 @@
 	} while (!frameQueue.empty());
 
 	if (frameQueueClosed) {
-		aio->queueWriteClose();
+            aio->queueWriteClose();
 	}
 }
 

Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp?rev=584143&r1=584142&r2=584143&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/AsynchIO.cpp Fri Oct 12 05:06:39 2007
@@ -157,7 +157,8 @@
 }
 
 void AsynchIO::queueWriteClose() {
-	queuedClose = true;
+    queuedClose = true;
+    DispatchHandle::rewatchWrite();
 }
 
 /** Return a queued buffer if there are enough



Mime
View raw message