diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 35ffc6b..7e45007 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -1,4 +1,5 @@ -find_package(Boost COMPONENTS date_time REQUIRED) +find_package(Boost COMPONENTS date_time filesystem system thread program_options REQUIRED) +find_package(Threads) include_directories( ${Boost_INLUDE_DIRS} ) @@ -10,9 +11,16 @@ add_library(slbu-lib SyslogMessage.cpp UDPWriter.cpp ) +target_link_libraries(slbu-lib + ${Boost_DATE_TIME_LIBRARY} + ${Boost_THREAD_LIBRARY} +) add_executable(syslog-bulk-uploader main.cpp) target_link_libraries(syslog-bulk-uploader + ${CMAKE_THREAD_LIBS_INIT} + ${Boost_SYSTEM_LIBRARY} + ${Boost_FILESYSTEM_LIBRARY} + ${Boost_PROGRAM_OPTIONS_LIBRARY} slbu-lib - ${Boost_DATE_TIME_LIBRARY} ) diff --git a/src/FileReader.h b/src/FileReader.h index b4fb795..404ec8a 100644 --- a/src/FileReader.h +++ b/src/FileReader.h @@ -27,11 +27,12 @@ SOFTWARE. #include #include "Reader.h" +#include "SyslogMessage.h" class FileReader : public Reader { public: - FileReader(const std::string filename) : _stream(std::ifstream(filename)) { + FileReader(const std::string& filename) : _stream(filename) { if (!_stream.is_open()) { throw "File not found: " + filename; } diff --git a/src/SyslogBulkUploader.cpp b/src/SyslogBulkUploader.cpp index e884d1b..85663de 100644 --- a/src/SyslogBulkUploader.cpp +++ b/src/SyslogBulkUploader.cpp @@ -25,9 +25,15 @@ SOFTWARE. #include "SyslogBulkUploader.h" #include "Reader.h" #include "Writer.h" +#include "FrequencyLimit.h" + +const size_t SyslogBulkUploader::DEFAULT_MPS; void SyslogBulkUploader::run() { + FrequencyLimit freqLimit(_mps); + while (auto msg = _reader.nextMessage()) { + freqLimit.tick(); _writer.sendMessage(msg); } } diff --git a/src/SyslogBulkUploader.h b/src/SyslogBulkUploader.h index dc0ec0a..9912faf 100644 --- a/src/SyslogBulkUploader.h +++ b/src/SyslogBulkUploader.h @@ -33,13 +33,16 @@ class Writer; class SyslogBulkUploader : boost::noncopyable { public: - SyslogBulkUploader(Reader& reader, Writer& writer) : _reader(reader), _writer(writer) { + SyslogBulkUploader(Reader& reader, Writer& writer, const size_t& mps = DEFAULT_MPS) : _reader(reader), + _writer(writer), _mps(mps) { }; void run(); private: + const static size_t DEFAULT_MPS = 1000; Reader& _reader; Writer& _writer; + const size_t _mps; }; #endif /* SYSLOGBULKUPLOADER_H */ diff --git a/src/UDPWriter.cpp b/src/UDPWriter.cpp index 81ac367..c9b8df1 100644 --- a/src/UDPWriter.cpp +++ b/src/UDPWriter.cpp @@ -29,7 +29,7 @@ SOFTWARE. using boost::asio::ip::udp; using namespace boost::asio; -UDPWriter::UDPWriter(const std::string& destination, const int port) { +UDPWriter::UDPWriter(const std::string& destination, const uint16_t port) { udp::resolver resolver(_ios); udp::resolver::query query(destination, std::to_string(port)); udp::resolver::iterator it = resolver.resolve(query); diff --git a/src/UDPWriter.h b/src/UDPWriter.h index 3a85e28..5ffb60c 100644 --- a/src/UDPWriter.h +++ b/src/UDPWriter.h @@ -32,7 +32,7 @@ SOFTWARE. class UDPWriter : public Writer { public: - UDPWriter(const std::string&, const int); + UDPWriter(const std::string&, const uint16_t); virtual void sendMessage(std::shared_ptr); diff --git a/src/main.cpp b/src/main.cpp index f4c34e5..e15dd28 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -21,10 +21,82 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ + #include +#include + +#include "SyslogBulkUploader.h" +#include "FileReader.h" +#include "UDPWriter.h" + +namespace po = boost::program_options; +using namespace std; + +size_t mps; +string dest; +uint16_t port; +vector files; int main(int argc, char** argv) { - std::cout << "Hello World!" << std::endl; + po::options_description desc("Supported options"); + desc.add_options() + ("help,h", "display help message") + ("version,v", "display version information") + ("mps,m", po::value(&mps)->default_value(1000), "try to maintain this rate of messages-per-second") + ("dest,d", po::value(&dest), "destination host name") + ("port,p", po::value(&port)->default_value(514), "destination port") + ("files,f", po::value> (&files), "input file(s)") + ; + po::positional_options_description pos; + pos.add("files", -1); + po::variables_map vars; + + try { + po::store(po::command_line_parser(argc, argv).options(desc).positional(pos).run(), vars); + } catch (exception& ex) { + cout << "Error parsing command line: " << ex.what() << endl; + return -1; + } + po::notify(vars); + + if (vars.count("help")) { + desc.print(std::cout); + return 0; + } + + if (vars.count("version")) { + cout << "VERSION" << std::endl; + return 0; + } + + if (!vars.count("dest")) { + cout << "ERROR: You must specify destination hostname" << endl; + desc.print(cout); + return -1; + } + + if (!vars.count("files")) { + cout << "ERROR: You must specify at least one input file" << endl; + desc.print(cout); + return -1; + } + + UDPWriter w(dest, port); + + for (auto& file : files) { + try { + FileReader r(file); + SyslogBulkUploader uploader(r, w, mps); + cout << "Sending logs from " << file << " to udp://" << dest << ":" << port << " at a max rate of " << + mps << " messages per second ... "; + uploader.run(); + cout << "done" << endl; + } catch (string& ex) { + cout << ex << endl; + } catch (exception& ex) { + cout << ex.what() << endl; + } + } } diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 3d38508..946b5c0 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -14,6 +14,7 @@ target_link_libraries(SyslogBulkUploaderTests ${Boost_FILESYSTEM_LIBRARY} ${Boost_SYSTEM_LIBRARY} ${CMAKE_THREAD_LIBS_INIT} + ${Boost_THREAD_LIBRARY} ) add_executable(SyslogMessageTests SyslogMessageTests.cpp) @@ -36,5 +37,6 @@ target_link_libraries(FrequencyLimitTests ${Boost_UNIT_TEST_FRAMEWORK_LIBRARY} ${Boost_DATE_TIME_LIBRARY} ${Boost_THREAD_LIBRARY} + ${Boost_SYSTEM_LIBRARY} )