Merge pull request #3 from MO-RISE/message_splitter

Reverting to a structure where an assembler is used to ensure a singl…

CMakeLists.txt

@@ -69,7 +69,7 @@ add_dependencies(${PROJECT_NAME} generate_message_specifications)
 CPMAddPackage("gh:onqtam/doctest#2.4.0")
 
 enable_testing()
-add_executable(${PROJECT_NAME}-tester ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_nmea_2000_assembler.cpp)
+add_executable(${PROJECT_NAME}-tester ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_assembler.cpp)
 target_include_directories(${PROJECT_NAME}-tester 
   PRIVATE 
   ${CMAKE_CURRENT_SOURCE_DIR}/src

README.md

@@ -2,7 +2,9 @@
 
 Copyright 2021 RISE Research Institute of Sweden - Maritime Operations. Licensed under the Apache License Version 2.0. For details, please contact Fredrik Olsson (fredrik.x.olsson(at)ri.se).
 
-A [libcluon](https://github.com/chrberger/libcluon)-based microservice for eavesdropping on a NMEA2000 stream over either UDP or TCP. This software does not perform any parsing or validation of the N2k frames, merely acts as a one-way bridge to a libcluon group.
+A [libcluon](https://github.com/chrberger/libcluon)-based microservice for eavesdropping on a NMEA2000 stream over either UDP or TCP. This software does not perform any parsing or validation of the N2k frames, merely acts as a one-way bridge to a libcluon group. However, it is currently specifically developed for receiving data from a YDEN-02 device in the format:
+
+    04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF
 
 ## How do I get it?
 Each release of `cluon-nmea2000` is published as a docker image [here](https://github.com/orgs/MO-RISE/packages/container/package/cluon-nmea2000) and is publicly available.

src/assembler.hpp

@@ -0,0 +1,76 @@
+///  Copyright 2021 RISE Research Institute of Sweden - Maritime Operations
+///
+///  Licensed under the Apache License, Version 2.0 (the "License");
+///  you may not use this file except in compliance with the License.
+///  You may obtain a copy of the License at
+///
+///      http://www.apache.org/licenses/LICENSE-2.0
+///
+///  Unless required by applicable law or agreed to in writing, software
+///  distributed under the License is distributed on an "AS IS" BASIS,
+///  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+///  See the License for the specific language governing permissions and
+///  limitations under the License.
+#ifndef SRC_ASSEMBLER_HPP_
+#define SRC_ASSEMBLER_HPP_
+
+#include <chrono>  // NOLINT
+#include <functional>
+#include <sstream>
+#include <string>
+#include <utility>
+
+#include "cluon/stringtoolbox.hpp"
+
+class Assembler {
+  std::string remainder_;
+  std::function<void(std::string &&, std::chrono::system_clock::time_point &&)>
+      delegate_{};
+
+ public:
+  Assembler(std::function<void(std::string &&,
+                               std::chrono::system_clock::time_point &&)>
+                delegate)
+      : delegate_(delegate) {}
+
+  void operator()(const std::string &&data,
+                  std::chrono::system_clock::time_point &&tp) {
+    // Get total buffered data
+    std::stringstream buffer{remainder_ + std::move(data)};
+
+    std::chrono::system_clock::time_point time_stamp{std::move(tp)};
+
+    // Handle any fully received lines
+    std::string line;
+    while (std::getline(buffer, line, '\n')) {
+      if (buffer.eof()) {
+        // If we run out of buffer, we dont have a full line and bail
+        // accordingly
+        break;
+      }
+
+      // Extract identifier and data
+      // Trim for any trailing whitespace
+      line.erase(line.find_last_not_of(" \n\r\t") + 1);
+      auto parts = stringtoolbox::split(line, ' ');
+
+      // Simple validation
+      // Example dataset
+      // 04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF
+      // 04:54:52.151 R 15FD070E 36 C2 9F 73 FF 7F FB 03
+      // 04:54:52.179 R 09F1120B B1 07 F2 00 00 FF 7F FD
+      if (parts.size() == 11) {
+        std::string identifier = parts[2];
+        std::string data;
+        for (auto it = parts.begin() + 3; it != parts.end(); ++it) {
+          data += *it;
+        }
+        delegate_(identifier + " " + data, std::move(time_stamp));
+      }
+    }
+    // Make sure we save any remaining characters for next incoming
+    remainder_ = line;
+  }
+};
+
+#endif  // SRC_ASSEMBLER_HPP_"

src/main.cpp

@@ -16,7 +16,7 @@
 #include <optional>
 
 #include "CLI/CLI.hpp"
-#include "YDEN-02_extractor.hpp"
+#include "assembler.hpp"
 #include "cluon/Envelope.hpp"
 #include "cluon/OD4Session.hpp"
 #include "cluon/TCPConnection.hpp"
@@ -45,28 +45,25 @@ auto main(int argc, char **argv) -> int {
   app.callback([&]() {
     cluon::OD4Session od4{cid};
 
+    Assembler assembler{
+        [&](std::string &&d, std::chrono::system_clock::time_point &&tp) {
+          auto timestamp = cluon::time::convert(tp);
+          memo::raw::NMEA2000 m;
+          m.data(d);
+          od4.send(m, timestamp, id);
+          if (verbose) {
+            std::cout << d << std::endl;
+          }
+        }};
+
     if (is_UDP) {
       // Setup a connection to an UDP source with incoming NMEA2000
       // messages
       cluon::UDPReceiver connection{
           address, port,
           [&](std::string &&d, std::string && /*from*/,
               std::chrono::system_clock::time_point &&tp) noexcept {
-            auto data = extract_N2k_frame(d);
-            if (data.empty()) {
-              return;
-            }
-
-            auto timestamp = cluon::time::convert(tp);
-
-            // Publish to OD4 session
-            memo::raw::NMEA2000 m;
-            m.data(data);
-            od4.send(m, timestamp, id);
-
-            if (verbose) {
-              std::cout << data << std::endl;
-            }
+            assembler(std::move(d), std::move(tp));
           }};
 
       using namespace std::literals::chrono_literals;  // NOLINT
@@ -79,21 +76,7 @@ auto main(int argc, char **argv) -> int {
           address, port,
           [&](std::string &&d,
               std::chrono::system_clock::time_point &&tp) noexcept {
-            auto data = extract_N2k_frame(d);
-            if (data.empty()) {
-              return;
-            }
-
-            auto timestamp = cluon::time::convert(tp);
-
-            // Publish to OD4 session
-            memo::raw::NMEA2000 m;
-            m.data(data);
-            od4.send(m, timestamp, id);
-
-            if (verbose) {
-              std::cout << data << std::endl;
-            }
+            assembler(std::move(d), std::move(tp));
           },
           [&argv]() {
             std::cerr << "[" << argv[0] << "] Connection lost." << std::endl;

tests/test_assembler.cpp

@@ -0,0 +1,139 @@
+///  Copyright 2021 RISE Research Institute of Sweden - Maritime Operations
+///
+///  Licensed under the Apache License, Version 2.0 (the "License");
+///  you may not use this file except in compliance with the License.
+///  You may obtain a copy of the License at
+///
+///      http://www.apache.org/licenses/LICENSE-2.0
+///
+///  Unless required by applicable law or agreed to in writing, software
+///  distributed under the License is distributed on an "AS IS" BASIS,
+///  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+///  See the License for the specific language governing permissions and
+///  limitations under the License.
+#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
+#include <doctest/doctest.h>
+
+#include <chrono>  // NOLINT
+#include <string>
+#include <vector>
+
+#include "assembler.hpp"
+
+TEST_CASE("Test Assembler with empty payload.") {
+  std::string output;
+
+  std::string DATA;
+
+  Assembler a{[&output](std::string&& line,
+                        std::chrono::system_clock::time_point&& timestamp) {
+    output = std::move(line);
+  }};
+  a(std::move(DATA), std::chrono::system_clock::time_point());
+
+  REQUIRE(output.empty());
+}
+
+TEST_CASE("Test Assembler with payload of wrong format") {
+  size_t call_count{0};
+  std::string output;
+
+  std::string DATA{"skfvdalfdadnsldasdc\nsakhbsacdsad\nahsdvds\nadjd"};
+
+  Assembler a{[&output, &call_count](
+                  std::string&& frame,
+                  std::chrono::system_clock::time_point&& timestamp) {
+    output = std::move(frame);
+    call_count++;
+  }};
+  a(std::move(DATA), std::chrono::system_clock::time_point());
+
+  REQUIRE(output.empty());
+  REQUIRE_EQ(call_count, 0);
+}
+
+TEST_CASE("Test Assembler with single correct sentence") {
+  size_t call_count{0};
+  std::string output;
+
+  std::string DATA{"04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF\r\n"};
+
+  Assembler a{[&output, &call_count](
+                  std::string&& frame,
+                  std::chrono::system_clock::time_point&& timestamp) {
+    output = std::move(frame);
+    call_count++;
+  }};
+  a(std::move(DATA), std::chrono::system_clock::time_point());
+
+  REQUIRE_EQ(output, "15FD080E 3600029F73FFFFFF");
+  REQUIRE_EQ(call_count, 1);
+}
+
+TEST_CASE(
+    "Test Assembler with single correct sentence inbetween "
+    "scramble") {
+  size_t call_count{0};
+  std::string output;
+
+  std::string DATA{"04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF\r\n"};
+
+  std::string scramble{"dadnsldasdc\n"};
+
+  Assembler a{[&output, &call_count](
+                  std::string&& frame,
+                  std::chrono::system_clock::time_point&& timestamp) {
+    output = std::move(frame);
+    call_count++;
+  }};
+  a(scramble + DATA + scramble, std::chrono::system_clock::time_point());
+
+  REQUIRE_EQ(output, "15FD080E 3600029F73FFFFFF");
+  REQUIRE_EQ(call_count, 1);
+}
+
+TEST_CASE("Test Assembler with multiple sentences") {
+  size_t call_count{0};
+  std::vector<std::string> output;
+
+  std::string DATA1{"04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF\r\n"};
+  std::string DATA2{"04:54:52.151 R 15FD070E 36 C2 9F 73 FF 7F FB 03\r\n"};
+  std::string DATA3{"04:54:52.179 R 09F1120B B1 07 F2 00 00 FF 7F FD\r\n"};
+
+  Assembler a{[&output, &call_count](
+                  std::string&& frame,
+                  std::chrono::system_clock::time_point&& timestamp) {
+    output.push_back(frame);
+    call_count++;
+  }};
+  a(std::move(DATA1), std::chrono::system_clock::time_point());
+  a(std::move(DATA2), std::chrono::system_clock::time_point());
+  a(std::move(DATA3), std::chrono::system_clock::time_point());
+
+  REQUIRE_EQ(call_count, 3);
+  REQUIRE_EQ(output[0], "15FD080E 3600029F73FFFFFF");
+  REQUIRE_EQ(output[1], "15FD070E 36C29F73FF7FFB03");
+  REQUIRE_EQ(output[2], "09F1120B B107F20000FF7FFD");
+}
+
+TEST_CASE("Test Assembler with split sentence") {
+  size_t call_count{0};
+  std::vector<std::string> output;
+
+  std::string DATA{"04:54:52.150 R 15FD080E 36 00 02 9F 73 FF FF FF\r\n"};
+
+  std::string input =
+      DATA.substr(13, DATA.size()) + DATA + DATA + DATA.substr(0, 17);
+
+  Assembler a{[&output, &call_count](
+                  std::string&& frame,
+                  std::chrono::system_clock::time_point&& timestamp) {
+    output.push_back(frame);
+    call_count++;
+  }};
+  a(std::move(input), std::chrono::system_clock::time_point());
+
+  REQUIRE_EQ(call_count, 2);
+  REQUIRE_EQ(output[0], "15FD080E 3600029F73FFFFFF");
+  REQUIRE_EQ(output[1], "15FD080E 3600029F73FFFFFF");
+}