fixed issues that caused reading data to be broken

This commit is contained in:
Emiel Bruijntjes 2016-06-15 14:50:33 -04:00
parent 99abd157e5
commit 3d9b5b444b
7 changed files with 33 additions and 53 deletions

1
src/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
*.d

View File

@ -1,6 +1,6 @@
CPP = g++ CPP = g++
RM = rm -f RM = rm -f
CPPFLAGS = -Wall -c -I. -std=c++11 CPPFLAGS = -Wall -c -I. -std=c++11 -MD
LD = g++ LD = g++
LD_FLAGS = -Wall -shared LD_FLAGS = -Wall -shared
SHARED_LIB = lib$(LIBRARY_NAME).so.$(VERSION) SHARED_LIB = lib$(LIBRARY_NAME).so.$(VERSION)
@ -8,6 +8,7 @@ STATIC_LIB = lib$(LIBRARY_NAME).a.$(VERSION)
SOURCES = $(wildcard *.cpp) SOURCES = $(wildcard *.cpp)
SHARED_OBJECTS = $(SOURCES:%.cpp=%.o) SHARED_OBJECTS = $(SOURCES:%.cpp=%.o)
STATIC_OBJECTS = $(SOURCES:%.cpp=%.s.o) STATIC_OBJECTS = $(SOURCES:%.cpp=%.s.o)
DEPENDENCIES = $(SOURCES:%.cpp=%.d)
PURE_SHARED_OBJECTS = $(filter-out tcpconnection.o, $(SOURCES:%.cpp=%.o)) PURE_SHARED_OBJECTS = $(filter-out tcpconnection.o, $(SOURCES:%.cpp=%.o))
PURE_STATIC_OBJECTS = $(filter-out tcpconnection.s.o, $(SOURCES:%.cpp=%.s.o)) PURE_STATIC_OBJECTS = $(filter-out tcpconnection.s.o, $(SOURCES:%.cpp=%.s.o))
@ -18,6 +19,7 @@ else
SONAMEPARAMETER = -soname SONAMEPARAMETER = -soname
endif endif
-include ${DEPENDENCIES}
all: CPPFLAGS += -g all: CPPFLAGS += -g
all: LD_FLAGS += -g all: LD_FLAGS += -g

View File

@ -3,7 +3,7 @@
* *
* Implementation of an AMQP connection * Implementation of an AMQP connection
* *
* @copyright 2014 Copernica BV * @copyright 2014 - 2016 Copernica BV
*/ */
#include "includes.h" #include "includes.h"
#include "protocolheaderframe.h" #include "protocolheaderframe.h"

View File

@ -86,8 +86,9 @@ namespace AMQP {
*/ */
ReceivedFrame::ReceivedFrame(const Buffer &buffer, uint32_t max) : _buffer(buffer) ReceivedFrame::ReceivedFrame(const Buffer &buffer, uint32_t max) : _buffer(buffer)
{ {
// we need enough room for type, channel, the payload size and the end-of-frame byte // we need enough room for type, channel, the payload size,
if (buffer.size() < 8) return; // the the end-of-frame byte is not yet necessary
if (buffer.size() < 7) return;
// get the information // get the information
_type = nextUint8(); _type = nextUint8();
@ -98,18 +99,13 @@ ReceivedFrame::ReceivedFrame(const Buffer &buffer, uint32_t max) : _buffer(buffe
if (max > 0 && _payloadSize > max - 8) throw ProtocolException("frame size exceeded"); if (max > 0 && _payloadSize > max - 8) throw ProtocolException("frame size exceeded");
// check if the buffer is big enough to contain all data // check if the buffer is big enough to contain all data
if (buffer.size() >= _payloadSize + 8) if (!complete()) return;
{
// buffer is big enough, check for a valid end-of-frame marker // buffer is big enough, check for a valid end-of-frame marker
if ((int)buffer.byte(_payloadSize+7) != END_OF_FRAME) throw ProtocolException("invalid end of frame marker"); if ((int)buffer.byte(_payloadSize+7) == END_OF_FRAME) return;
}
else // the frame is invalid because it does not end with the expected char
{ throw ProtocolException("invalid end of frame marker");
// frame is not yet valid
_type = TYPE_INVALID;
_channel = 0;
_payloadSize = 0;
}
} }
/** /**
@ -127,7 +123,7 @@ bool ReceivedFrame::header() const
*/ */
bool ReceivedFrame::complete() const bool ReceivedFrame::complete() const
{ {
return _type != TYPE_INVALID; return _buffer.size() >= _payloadSize + 8;
} }
/** /**

View File

@ -161,7 +161,7 @@ public:
buffer.shrink(processed); buffer.shrink(processed);
// restore the buffer as member // restore the buffer as member
std::swap(_in, buffer); _in = std::move(buffer);
} }
// keep same object // keep same object

View File

@ -22,19 +22,13 @@ namespace AMQP {
*/ */
class TcpInBuffer : public ByteBuffer class TcpInBuffer : public ByteBuffer
{ {
private:
/**
* Number of bytes already filled
* @var size_t
*/
size_t _filled = 0;
public: public:
/** /**
* Constructor * Constructor
* @param size initial size * Note that we pass 0 to the constructor because the buffer seems to be empty
* @param size initial size to allocated
*/ */
TcpInBuffer(size_t size) : ByteBuffer((char *)malloc(size), size) {} TcpInBuffer(size_t size) : ByteBuffer((char *)malloc(size), 0) {}
/** /**
* No copy'ing * No copy'ing
@ -46,11 +40,7 @@ public:
* Move constructor * Move constructor
* @param that * @param that
*/ */
TcpInBuffer(TcpInBuffer &&that) : ByteBuffer(std::move(that)), _filled(that._filled) TcpInBuffer(TcpInBuffer &&that) : ByteBuffer(std::move(that)) {}
{
// reset other object
that._filled = 0;
}
/** /**
* Destructor * Destructor
@ -70,14 +60,8 @@ public:
// skip self-assignment // skip self-assignment
if (this == &that) return *this; if (this == &that) return *this;
// copy the filled paramteer
_filled = that._filled;
// reset other object
that._filled = 0;
// call base // call base
ByteBuffer::operator=(std::move(*this)); ByteBuffer::operator=(std::move(that));
// done // done
return *this; return *this;
@ -91,10 +75,7 @@ public:
{ {
// update data // update data
_data = (char *)realloc((void *)_data, size); _data = (char *)realloc((void *)_data, size);
}
// update size
_size = size;
}
/** /**
* Receive data from a socket * Receive data from a socket
@ -115,13 +96,13 @@ public:
if (available == 0) available = 1; if (available == 0) available = 1;
// number of bytes to read // number of bytes to read
size_t bytes = std::min(expected, available); size_t bytes = std::min((uint32_t)(expected - _size), available);
// read data into the buffer // read data into the buffer
auto result = read(socket, (void *)(_data + _filled), bytes); auto result = read(socket, (void *)(_data + _size), bytes);
// update total buffer size // update total buffer size
if (result > 0) _filled += result; if (result > 0) _size += result;
// done // done
return result; return result;
@ -133,8 +114,8 @@ public:
*/ */
void shrink(size_t size) void shrink(size_t size)
{ {
// update filled bytes // update size
_filled -= size; _size -= size;
} }
}; };

View File

@ -4,7 +4,7 @@
* Test program to check AMQP functionality based on LibEV * Test program to check AMQP functionality based on LibEV
* *
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com> * @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2015 Copernica BV * @copyright 2015 - 2016 Copernica BV
*/ */
/** /**
@ -27,7 +27,7 @@ int main()
AMQP::LibEvHandler handler(loop); AMQP::LibEvHandler handler(loop);
// make a connection // make a connection
AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://localhost/")); AMQP::TcpConnection connection(&handler, AMQP::Address("amqp://guest:guest@localhost/"));
// we need a channel too // we need a channel too
AMQP::TcpChannel channel(&connection); AMQP::TcpChannel channel(&connection);