Bugfixes for the BufferedStream class.

This commit is contained in:
Gunnar Beutner 2013-04-19 14:47:41 +02:00
parent 09d97c551e
commit fe4fd67405
4 changed files with 66 additions and 25 deletions

View File

@ -27,14 +27,33 @@
using namespace icinga;
BufferedStream::BufferedStream(const Stream::Ptr& innerStream)
: m_InnerStream(innerStream), m_RecvQ(boost::make_shared<FIFO>()), m_SendQ(boost::make_shared<FIFO>()),
m_Exception(), m_Blocking(true)
: m_InnerStream(innerStream), m_Stopped(false),
m_RecvQ(boost::make_shared<FIFO>()), m_SendQ(boost::make_shared<FIFO>()),
m_Blocking(true), m_Exception()
{
boost::thread readThread(boost::bind(&BufferedStream::ReadThreadProc, this));
readThread.detach();
m_ReadThread = boost::thread(boost::bind(&BufferedStream::ReadThreadProc, this));
m_WriteThread = boost::thread(boost::bind(&BufferedStream::WriteThreadProc, this));
}
boost::thread writeThread(boost::bind(&BufferedStream::WriteThreadProc, this));
writeThread.detach();
BufferedStream::~BufferedStream(void)
{
{
boost::mutex::scoped_lock lock(m_Mutex);
m_Stopped = true;
}
m_InnerStream->Close();
{
boost::mutex::scoped_lock lock(m_Mutex);
m_ReadCV.notify_all();
m_WriteCV.notify_all();
}
m_ReadThread.join();
m_WriteThread.join();
}
void BufferedStream::ReadThreadProc(void)
@ -51,6 +70,9 @@ void BufferedStream::ReadThreadProc(void)
boost::mutex::scoped_lock lock(m_Mutex);
m_RecvQ->Write(buffer, rc);
m_ReadCV.notify_all();
if (m_Stopped)
break;
}
} catch (const std::exception& ex) {
{
@ -75,9 +97,12 @@ void BufferedStream::WriteThreadProc(void)
{
boost::mutex::scoped_lock lock(m_Mutex);
while (m_SendQ->GetAvailableBytes() == 0)
while (m_SendQ->GetAvailableBytes() == 0 && !m_Stopped)
m_WriteCV.wait(lock);
if (m_Stopped)
break;
rc = m_SendQ->Read(buffer, sizeof(buffer));
}
@ -152,7 +177,7 @@ void BufferedStream::InternalWaitReadable(size_t count, boost::mutex::scoped_loc
m_ReadCV.wait(lock);
}
void BufferedStream::WaitWritable(size_t count)
void BufferedStream::WaitWritable(size_t)
{ /* Nothing to do here. */ }
void BufferedStream::MakeNonBlocking(void)
@ -161,4 +186,3 @@ void BufferedStream::MakeNonBlocking(void)
m_Blocking = false;
}

View File

@ -39,6 +39,7 @@ public:
typedef weak_ptr<BufferedStream> WeakPtr;
BufferedStream(const Stream::Ptr& innerStream);
~BufferedStream(void);
virtual size_t Read(void *buffer, size_t count);
virtual void Write(const void *buffer, size_t count);
@ -53,6 +54,8 @@ public:
private:
Stream::Ptr m_InnerStream;
bool m_Stopped;
FIFO::Ptr m_RecvQ;
FIFO::Ptr m_SendQ;
@ -67,6 +70,9 @@ private:
void ReadThreadProc(void);
void WriteThreadProc(void);
boost::thread m_ReadThread;
boost::thread m_WriteThread;
void InternalWaitReadable(size_t count, boost::mutex::scoped_lock& lock);
};

View File

@ -27,10 +27,18 @@ BOOST_AUTO_TEST_SUITE(base_match)
BOOST_AUTO_TEST_CASE(tolong)
{
BOOST_CHECK(Utility::Match("*", "hello"));
BOOST_CHECK(!Utility::Match("\\**", "hello"));
BOOST_CHECK(Utility::Match("\\**", "*ello"));
BOOST_CHECK(Utility::Match("?e*l?", "hello"));
BOOST_CHECK(Utility::Match("?e*l?", "helo"));
BOOST_CHECK(!Utility::Match("world", "hello"));
BOOST_CHECK(!Utility::Match("hee*", "hello"));
BOOST_CHECK(Utility::Match("he??o", "hello"));
BOOST_CHECK(Utility::Match("he?", "hel"));
BOOST_CHECK(Utility::Match("he*", "hello"));
BOOST_CHECK(Utility::Match("he*o", "heo"));
BOOST_CHECK(Utility::Match("he**o", "heo"));
BOOST_CHECK(Utility::Match("he**o", "hello"));
}
BOOST_AUTO_TEST_SUITE_END()

View File

@ -45,6 +45,7 @@
* And last but not least, '\?' and '\*' in `new_mask' now become one character.
*/
#if 0
int mmatch(const char *old_mask, const char *new_mask)
{
const char *m = old_mask;
@ -145,6 +146,7 @@ int mmatch(const char *old_mask, const char *new_mask)
}
}
}
#endif
/*
* Compare if a given string (name) matches the given
@ -261,6 +263,7 @@ break_while:
* Note that this new optimized alghoritm can *only* work in place.
*/
#if 0
char *collapse(char *pattern)
{
int star = 0;
@ -303,4 +306,4 @@ char *collapse(char *pattern)
};
return pattern;
}
#endif