Implement command line parsing; Implement messages-per-second limits in uploader; First playable version

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}
 )

src/FileReader.h

@@ -27,11 +27,12 @@ SOFTWARE.
 
 #include <fstream>
 #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;
         }

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);
     }
 }

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 */

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);

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<const SyslogMessage>);
 

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 <iostream>
+#include <boost/program_options.hpp>
+
+#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<string> 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<size_t>(&mps)->default_value(1000), "try to maintain this rate of messages-per-second")
+            ("dest,d", po::value<string>(&dest), "destination host name")
+            ("port,p", po::value<uint16_t>(&port)->default_value(514), "destination port")
+            ("files,f", po::value<vector < string >> (&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;
+        }
+    }
 }
 
 

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}
 )