1 files changed, 14 insertions, 54 deletions

diff --git a/apps/sm-serial-read.cpp b/apps/sm-serial-read.cpp
index d605ae4..955d262 100644
--- a/apps/sm-serial-read.cpp
+++ b/apps/sm-serial-read.cpp
@@ -4,7 +4,6 @@
 #include <chrono>
 #include <thread>
 #include <boost/asio/serial_port.hpp>
-#include <boost/regex.hpp>
 
 enum class Format {
     PLAIN,
@@ -60,74 +59,32 @@ string hostname = get_hostname();
 
 class port_handler {
 public:
-    port_handler(string device, serial_port &serial_port, shared_ptr<SampleOutputStream> stream) :
-            device(device), port(serial_port), stream(stream) {
+    port_handler(string device, serial_port &serial_port, shared_ptr<CsvParser> input) :
+            device(device), port(serial_port), input(input) {
     }
 
     void run() {
         auto packet = make_shared<vector<uint8_t>>(1024);
 
         while (port.is_open()) {
-            size_t some = port.read_some(buffer);
-            for (int i = 0; i < some; i++) {
-                uint8_t b = data[i];
+            std::size_t some = port.read_some(buffer);
 
-                if (b == packet_delimiter) {
-                    on_packet(packet);
-                    packet = make_shared<vector<uint8_t>>(1024);
-                } else {
-                    packet->emplace_back(b);
-                }
-            }
+            mutable_buffers_1 chunk = boost::asio::buffer(data, some);
+            input->process(chunk);
         }
 
         cerr << "port closed" << endl;
     }
 
-    void on_packet(shared_ptr<vector<uint8_t>> packet) {
-        auto timestamp = std::chrono::system_clock::now().time_since_epoch().count();
-        auto s = std::string((char *) packet->data(), packet->size());
-        cerr << "packet: " << s << endl;
-
-        static const boost::regex e("([a-zA-Z0-9]+)=([0-9]+)");
-
-        std::string::const_iterator start = s.begin();
-        std::string::const_iterator end = s.end();
-        boost::match_results<std::string::const_iterator> what;
-        boost::match_flag_type flags = boost::match_default;
-
-        while (regex_search(start, end, what, e, flags)) {
-            auto sensor = static_cast<string>(what[1]);
-            auto value = static_cast<string>(what[2]);
-            start = what[0].second;
-
-            static const string device_type = "serial";
-            map<string, string> values;
-            values["hostname"] = hostname;
-            values["device_type"] = device_type;
-            values["device"] = device;
-            values["timestamp"] = to_string(timestamp);
-            values["sensor"] = sensor;
-            values["value"] = value;
-
-//            cerr << sensor << " => " << value << endl;
-
-            stream->write(values);
-
-            flags |= boost::match_prev_avail;
-            flags |= boost::match_not_bob;
-        }
-    }
 
 private:
     static const size_t size = 1024;
-    static const uint8_t packet_delimiter = '\n';
     string device;
     serial_port &port;
     uint8_t data[size];
     mutable_buffers_1 buffer = boost::asio::buffer(data, size);
 
-    shared_ptr<SampleOutputStream> stream;
+    shared_ptr<CsvParser> input;
 };
 
 class sm_serial_read : public app {
@@ -171,20 +128,23 @@ public:
                 "sensor",
                 "value"
         });
-        shared_ptr <SampleOutputStream> sampleStream;
+
+        shared_ptr<SampleOutputStream> output;
 
         if (format == Format::JSON) {
-            sampleStream = make_shared<JsonSampleOutputStream>(cout, field_names);
+            output = make_shared<JsonSampleOutputStream>(cout, field_names);
         } else if (format == Format::SQL) {
-            sampleStream = make_shared<SqlSampleOutputStream>(cout, field_names);
+            output = make_shared<SqlSampleOutputStream>(cout, field_names);
         } else if (format == Format::PLAIN) {
-            sampleStream = make_shared<CsvSampleOutputStream>(cout, field_names);
+            output = make_shared<CsvSampleOutputStream>(cout, field_names);
         } else {
             cerr << "Unsupported format: " << boost::lexical_cast<string>(format) << endl;
             return EXIT_FAILURE;
         }
 
-        port_handler(port_name, port, sampleStream).run();
+        shared_ptr<CsvParser> input = make_shared<CsvParser>(output);
+
+        port_handler(port_name, port, input).run();
 
         return EXIT_SUCCESS;
     }