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
