138 lines
2.4 KiB
C++
138 lines
2.4 KiB
C++
#include "CInputPacketStream.h"
|
|
#include "CLock.h"
|
|
|
|
//
|
|
// CInputPacketStream
|
|
//
|
|
|
|
CInputPacketStream::CInputPacketStream(IInputStream* stream, bool adopt) :
|
|
CInputStreamFilter(stream, adopt),
|
|
m_mutex(),
|
|
m_size(0),
|
|
m_buffer(&m_mutex, NULL)
|
|
{
|
|
// do nothing
|
|
}
|
|
|
|
CInputPacketStream::~CInputPacketStream()
|
|
{
|
|
// do nothing
|
|
}
|
|
|
|
void
|
|
CInputPacketStream::close()
|
|
{
|
|
getStream()->close();
|
|
}
|
|
|
|
UInt32
|
|
CInputPacketStream::read(void* buffer, UInt32 n)
|
|
{
|
|
CLock lock(&m_mutex);
|
|
|
|
// wait for entire message to be read. return immediately if
|
|
// stream hungup.
|
|
if (!waitForFullMessage()) {
|
|
return 0;
|
|
}
|
|
|
|
// limit number of bytes to read to the number of bytes left in the
|
|
// current message.
|
|
if (n > m_size) {
|
|
n = m_size;
|
|
}
|
|
|
|
// now read from our buffer
|
|
n = m_buffer.readNoLock(buffer, n);
|
|
assert(n <= m_size);
|
|
m_size -= n;
|
|
|
|
return n;
|
|
}
|
|
|
|
UInt32
|
|
CInputPacketStream::getSize() const
|
|
{
|
|
CLock lock(&m_mutex);
|
|
return getSizeNoLock();
|
|
}
|
|
|
|
UInt32
|
|
CInputPacketStream::getSizeNoLock() const
|
|
{
|
|
while (!hasFullMessage() && getStream()->getSize() > 0) {
|
|
// read more data
|
|
if (!getMoreMessage()) {
|
|
// stream hungup
|
|
return false;
|
|
}
|
|
}
|
|
|
|
return m_size;
|
|
}
|
|
|
|
bool
|
|
CInputPacketStream::waitForFullMessage() const
|
|
{
|
|
while (!hasFullMessage()) {
|
|
// read more data
|
|
if (!getMoreMessage()) {
|
|
// stream hungup
|
|
return false;
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
bool
|
|
CInputPacketStream::getMoreMessage() const
|
|
{
|
|
// read more data
|
|
char buffer[4096];
|
|
UInt32 n = getStream()->read(buffer, sizeof(buffer));
|
|
|
|
// return if stream hungup
|
|
if (n == 0) {
|
|
m_buffer.hangup();
|
|
return false;
|
|
}
|
|
|
|
// append to our buffer
|
|
m_buffer.write(buffer, n);
|
|
|
|
return true;
|
|
}
|
|
|
|
bool
|
|
CInputPacketStream::hasFullMessage() const
|
|
{
|
|
// get payload length if we don't have it yet
|
|
if (m_size == 0) {
|
|
// check if length field has been read yet
|
|
if (m_buffer.getSizeNoLock() < 4) {
|
|
// not enough data for payload length
|
|
return false;
|
|
}
|
|
|
|
// save payload length
|
|
UInt8 buffer[4];
|
|
UInt32 n = m_buffer.readNoLock(buffer, sizeof(buffer));
|
|
assert(n == 4);
|
|
m_size = ((UInt32)buffer[0] << 24) |
|
|
((UInt32)buffer[1] << 16) |
|
|
((UInt32)buffer[2] << 8) |
|
|
(UInt32)buffer[3];
|
|
|
|
// if payload length is zero then discard null message
|
|
if (m_size == 0) {
|
|
return false;
|
|
}
|
|
}
|
|
assert(m_size > 0);
|
|
|
|
// we have the full message when we have at least m_size bytes in
|
|
// the buffer
|
|
return (m_buffer.getSizeNoLock() >= m_size);
|
|
}
|