Fix invalid handling of input filenames command line option; Print dot every second while sending messages to indicate activity
This commit is contained in:
+1
-1
@@ -39,7 +39,7 @@ set(CPACK_GENERATOR "RPM;DEB;TGZ")
|
|||||||
set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "${VER_NAME}")
|
set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "${VER_NAME}")
|
||||||
set(CPACK_PACKAGE_VENDOR "Marko Zivanovic <marko@zivanovic.in.rs>")
|
set(CPACK_PACKAGE_VENDOR "Marko Zivanovic <marko@zivanovic.in.rs>")
|
||||||
set(CPACK_PACKAGE_CONTACT "${CPACK_PACKAGE_VENDOR}")
|
set(CPACK_PACKAGE_CONTACT "${CPACK_PACKAGE_VENDOR}")
|
||||||
set(CPACK_PACKAGE_DESCRIPTION_FILE "${CMAKE_CURRENT_SOURCE_DIR}/../README.md")
|
set(CPACK_PACKAGE_DESCRIPTION_FILE "${CMAKE_CURRENT_SOURCE_DIR}/PackageDescription.txt")
|
||||||
set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Utility to send messages from file to remote syslog server")
|
set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Utility to send messages from file to remote syslog server")
|
||||||
set(CPACK_RESOURCE_FILE_LICENSE "${CMAKE_CURRENT_SOURCE_DIR}/../LICENSE")
|
set(CPACK_RESOURCE_FILE_LICENSE "${CMAKE_CURRENT_SOURCE_DIR}/../LICENSE")
|
||||||
set(CPACK_PACKAGE_VERSION "${SLBU_VER}")
|
set(CPACK_PACKAGE_VERSION "${SLBU_VER}")
|
||||||
|
|||||||
@@ -34,6 +34,10 @@ void SyslogBulkUploader::run() {
|
|||||||
|
|
||||||
while (auto msg = _reader.nextMessage()) {
|
while (auto msg = _reader.nextMessage()) {
|
||||||
freqLimit.tick();
|
freqLimit.tick();
|
||||||
|
if (_preSendCallback)
|
||||||
|
_preSendCallback(msg);
|
||||||
_writer.sendMessage(msg);
|
_writer.sendMessage(msg);
|
||||||
|
if (_postSendCallback)
|
||||||
|
_postSendCallback(msg);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,23 +26,38 @@ SOFTWARE.
|
|||||||
#define SYSLOGBULKUPLOADER_H
|
#define SYSLOGBULKUPLOADER_H
|
||||||
|
|
||||||
#include <boost/noncopyable.hpp>
|
#include <boost/noncopyable.hpp>
|
||||||
|
#include <functional>
|
||||||
|
#include <memory>
|
||||||
|
|
||||||
class Reader;
|
class Reader;
|
||||||
class Writer;
|
class Writer;
|
||||||
|
class SyslogMessage;
|
||||||
|
|
||||||
class SyslogBulkUploader : boost::noncopyable {
|
class SyslogBulkUploader : boost::noncopyable {
|
||||||
public:
|
public:
|
||||||
|
|
||||||
|
typedef std::function<void(std::shared_ptr<const SyslogMessage>)> Callback;
|
||||||
|
|
||||||
SyslogBulkUploader(Reader& reader, Writer& writer, const size_t& mps = DEFAULT_MPS) : _reader(reader),
|
SyslogBulkUploader(Reader& reader, Writer& writer, const size_t& mps = DEFAULT_MPS) : _reader(reader),
|
||||||
_writer(writer), _mps(mps) {
|
_writer(writer), _mps(mps) {
|
||||||
};
|
};
|
||||||
void run();
|
void run();
|
||||||
|
|
||||||
|
void setPreSendCallback(Callback cb) {
|
||||||
|
_preSendCallback = cb;
|
||||||
|
}
|
||||||
|
|
||||||
|
void setPostSendCallback(Callback cb) {
|
||||||
|
_postSendCallback = cb;
|
||||||
|
}
|
||||||
|
|
||||||
private:
|
private:
|
||||||
const static size_t DEFAULT_MPS = 1000;
|
const static size_t DEFAULT_MPS = 1000;
|
||||||
Reader& _reader;
|
Reader& _reader;
|
||||||
Writer& _writer;
|
Writer& _writer;
|
||||||
const size_t _mps;
|
const size_t _mps;
|
||||||
|
Callback _preSendCallback;
|
||||||
|
Callback _postSendCallback;
|
||||||
};
|
};
|
||||||
|
|
||||||
#endif /* SYSLOGBULKUPLOADER_H */
|
#endif /* SYSLOGBULKUPLOADER_H */
|
||||||
|
|||||||
+18
-4
@@ -33,12 +33,16 @@ SOFTWARE.
|
|||||||
|
|
||||||
namespace po = boost::program_options;
|
namespace po = boost::program_options;
|
||||||
using namespace std;
|
using namespace std;
|
||||||
|
using namespace boost::posix_time;
|
||||||
|
|
||||||
po::options_description desc("Supported options");
|
po::options_description desc("Supported options");
|
||||||
size_t mps;
|
size_t mps;
|
||||||
string dest;
|
string dest;
|
||||||
uint16_t port;
|
uint16_t port;
|
||||||
vector<string> files;
|
vector<string> files;
|
||||||
|
ptime start;
|
||||||
|
ptime lastPrint;
|
||||||
|
time_duration printInterval = seconds(1);
|
||||||
|
|
||||||
string version() {
|
string version() {
|
||||||
stringstream ss;
|
stringstream ss;
|
||||||
@@ -50,6 +54,14 @@ void help() {
|
|||||||
desc.print(cout);
|
desc.print(cout);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void preSendCallback(shared_ptr<const SyslogMessage>) {
|
||||||
|
ptime now = second_clock::local_time();
|
||||||
|
if (now - lastPrint > printInterval) {
|
||||||
|
cout << "." << flush;
|
||||||
|
lastPrint = now;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
int main(int argc, char** argv) {
|
int main(int argc, char** argv) {
|
||||||
desc.add_options()
|
desc.add_options()
|
||||||
("help,h", "display help message")
|
("help,h", "display help message")
|
||||||
@@ -60,7 +72,7 @@ int main(int argc, char** argv) {
|
|||||||
("file,f", po::value<vector < string >> (&files), "input file(s)")
|
("file,f", po::value<vector < string >> (&files), "input file(s)")
|
||||||
;
|
;
|
||||||
po::positional_options_description pos;
|
po::positional_options_description pos;
|
||||||
pos.add("files", -1);
|
pos.add("file", -1);
|
||||||
po::variables_map vars;
|
po::variables_map vars;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
@@ -87,7 +99,7 @@ int main(int argc, char** argv) {
|
|||||||
return -1;
|
return -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!vars.count("files")) {
|
if (!vars.count("file")) {
|
||||||
cout << "ERROR: You must specify at least one input file" << endl;
|
cout << "ERROR: You must specify at least one input file" << endl;
|
||||||
help();
|
help();
|
||||||
return -1;
|
return -1;
|
||||||
@@ -95,14 +107,16 @@ int main(int argc, char** argv) {
|
|||||||
|
|
||||||
UDPWriter w(dest, port);
|
UDPWriter w(dest, port);
|
||||||
|
|
||||||
|
lastPrint = start = second_clock::local_time();
|
||||||
for (auto& file : files) {
|
for (auto& file : files) {
|
||||||
try {
|
try {
|
||||||
FileReader r(file);
|
FileReader r(file);
|
||||||
SyslogBulkUploader uploader(r, w, mps);
|
SyslogBulkUploader uploader(r, w, mps);
|
||||||
|
uploader.setPostSendCallback(bind(preSendCallback, placeholders::_1));
|
||||||
cout << "Sending logs from " << file << " to udp://" << dest << ":" << port << " at a max rate of " <<
|
cout << "Sending logs from " << file << " to udp://" << dest << ":" << port << " at a max rate of " <<
|
||||||
mps << " messages per second ... " << flush;
|
mps << " messages per second " << flush;
|
||||||
uploader.run();
|
uploader.run();
|
||||||
cout << "done" << endl;
|
cout << " - DONE" << endl;
|
||||||
} catch (string& ex) {
|
} catch (string& ex) {
|
||||||
cout << ex << endl;
|
cout << ex << endl;
|
||||||
} catch (exception& ex) {
|
} catch (exception& ex) {
|
||||||
|
|||||||
Reference in New Issue
Block a user