generic-library/poco
1
0
Fork 0
mirror of https://github.com/pocoproject/poco.git synced 2025-10-27 02:53:10 +01:00
Code Issues Projects Releases Wiki Activity

#2787: add queueSize property to the AsyncChannel

Browse Source
This commit is contained in:
Günter Obiltschnig
2020-02-04 10:53:01 +01:00
parent 81624a269f
commit c4bdadffb3
2 changed files with 63 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
Foundation/src/AsyncChannel.cpp
View File

@@ -18,7 +18,10 @@
#include "Poco/Formatter.h"
#include "Poco/AutoPtr.h"
#include "Poco/LoggingRegistry.h"
#include "Poco/NumberParser.h"
#include "Poco/Exception.h"
#include "Poco/String.h"
#include "Poco/Format.h"
namespace Poco {
@@ -31,23 +34,23 @@ public:
_msg(msg)
{
}
~MessageNotification()
{
}
const Message& message() const
{
return _msg;
}
private:
Message _msg;
};
AsyncChannel::AsyncChannel(Channel::Ptr pChannel, Thread::Priority prio):
_pChannel(pChannel),
AsyncChannel::AsyncChannel(Channel::Ptr pChannel, Thread::Priority prio):
_pChannel(pChannel),
_thread("AsyncChannel")
{
_thread.setPriority(prio);
@@ -70,7 +73,7 @@ AsyncChannel::~AsyncChannel()
void AsyncChannel::setChannel(Channel::Ptr pChannel)
{
FastMutex::ScopedLock lock(_channelMutex);
_pChannel = pChannel;
}
@@ -94,10 +97,10 @@ void AsyncChannel::close()
if (_thread.isRunning())
{
while (!_queue.empty()) Thread::sleep(100);
do
do
{
_queue.wakeUpAll();
_queue.wakeUpAll();
}
while (!_thread.tryJoin(100));
}
@@ -106,6 +109,18 @@ void AsyncChannel::close()
void AsyncChannel::log(const Message& msg)
{
if (_queueSize != 0 && _queue.size() >= _queueSize)
{
++_dropCount;
return;
}
if (_dropCount != 0)
{
_queue.enqueueNotification(new MessageNotification(Message(msg, Poco::format("Dropped %z messages.", _dropCount))));
_dropCount = 0;
}
open();
_queue.enqueueNotification(new MessageNotification(msg));
@@ -115,11 +130,24 @@ void AsyncChannel::log(const Message& msg)
void AsyncChannel::setProperty(const std::string& name, const std::string& value)
{
if (name == "channel")
{
setChannel(LoggingRegistry::defaultRegistry().channelForName(value));
}
else if (name == "priority")
{
setPriority(value);
}
else if (name == "queueSize")
{
if (Poco::icompare(value, "none") == 0 || Poco::icompare(value, "unlimited") == 0 || value.empty())
_queueSize = 0;
else
_queueSize = Poco::NumberParser::parseUnsigned(value);
}
else
{
Channel::setProperty(name, value);
}
}
@@ -137,12 +165,12 @@ void AsyncChannel::run()
nf = _queue.waitDequeueNotification();
}
}
void AsyncChannel::setPriority(const std::string& value)
{
Thread::Priority prio = Thread::PRIO_NORMAL;
if (value == "lowest")
prio = Thread::PRIO_LOWEST;
else if (value == "low")
@@ -155,7 +183,7 @@ void AsyncChannel::setPriority(const std::string& value)
prio = Thread::PRIO_HIGHEST;
else
throw InvalidArgumentException("thread priority", value);
_thread.setPriority(prio);
}