Merge pull request #4 from RISE-MO/tcp-support

Adding tcp support

README.md

@@ -2,8 +2,8 @@
 
 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 NMEA0183 stream over (UDP). This software does not perform any parsing of the NMEA sentences, merely assembles any fragmented messages into full sentences. It can be run in two modes:
-* `gather`, connect to an UDP stream of incoming NMEA0183 sentences and either:
+A [libcluon](https://github.com/chrberger/libcluon)-based microservice for eavesdropping on a NMEA0183 stream over either UDP or TCP. This software does not perform any parsing of the NMEA sentences, merely assembles any fragmented messages into full sentences. It can be run in two modes:
+* `gather`, connect to a stream of incoming NMEA0183 sentences and either:
   * publish to an OD4 session, or
   * log directly to disk (`--standalone`)
 * `log`, listen to an OD4 session for raw NMEA0183 messages from other `gatherers`  and dump these to an aggregated log file on disk
@@ -14,25 +14,25 @@ Each release of `cluon-nmea0183` is published as a docker image [here](https://g
 Can also be used as a standalone commandline tool. No pre-built binaries are, however, provided for this purpose.
 
 ## Example docker-compose setup
-The example below showcases a setup with two gatherers (listening on two separate UDP streams) and one logger that aggregates published messages from the gatherers into a single file.
+The example below showcases a setup with two gatherers (listening on two separate stream (one UDP and one TCP)) and one logger that aggregates published messages from the gatherers into a single file.
 ```yaml
 version: '2'
 services:    
     gatherer_1:
         container_name: cluon-nmea0183-gatherer-1
-        image: ghcr.io/rise-mo/cluon-nmea0183:v0.1.0
+        image: ghcr.io/rise-mo/cluon-nmea0183:v0.2.0
         restart: on-failure
         network_mode: "host"
-        command: "--cid 111 --id 1 gather -a 255.255.255.255 -p 1456"
+        command: "--cid 111 --id 1 gather --udp -a 255.255.255.255 -p 1456"
     gatherer_2:
         container_name: cluon-nmea0183-gatherer-2
-        image: ghcr.io/rise-mo/cluon-nmea0183:v0.1.0
+        image: ghcr.io/rise-mo/cluon-nmea0183:v0.2.0
         restart: on-failure
         network_mode: "host"
-        command: "--cid 111 --id 2 gather -a 239.192.0.3 -p 60003"
+        command: "--cid 111 --id 2 gather -a 171.31.16.42 -p 6002"
     logger:
         container_name: cluon-nmea0183-logger
-        image: ghcr.io/rise-mo/cluon-nmea0183:v0.1.0
+        image: ghcr.io/rise-mo/cluon-nmea0183:v0.2.0
         restart: on-failure
         network_mode: "host"
         volumes:

src/main.cpp

@@ -19,6 +19,7 @@
 #include "NMEA0183_assembler.hpp"
 #include "cluon/Envelope.hpp"
 #include "cluon/OD4Session.hpp"
+#include "cluon/TCPConnection.hpp"
 #include "cluon/UDPReceiver.hpp"
 #include "risemo-message-set.hpp"
 #include "spdlog/sinks/daily_file_sink.h"
@@ -53,6 +54,9 @@ auto main(int argc, char **argv) -> int {
   uint16_t port;
   gather->add_option("-p,--port", port, "Port number to connect to")
       ->required();
+  bool is_UDP = false;
+  gather->add_flag("--udp", is_UDP,
+                   "If application should use UDP to connect, default is TCP");
   bool standalone = false;
   gather->add_flag("--standalone", standalone,
                    "If application should be run in standalone mode, i.e. "
@@ -96,18 +100,39 @@ auto main(int argc, char **argv) -> int {
     // And finally use this in a SentenceAssembler
     NMEA0183SentenceAssembler assembler(sentence_handler);
 
-    // Setup a connection to an UDP source with incoming NMEA0183
-    // messages
-    cluon::UDPReceiver connection{
-        address, port,
-        [&assembler](std::string &&d, std::string && /*from*/,
-                     std::chrono::system_clock::time_point &&tp) noexcept {
-          assembler(d, std::move(tp));
-        }};
-
-    using namespace std::literals::chrono_literals;  // NOLINT
-    while (connection.isRunning()) {
-      std::this_thread::sleep_for(1s);
+    if (is_UDP) {
+      // Setup a connection to an UDP source with incoming NMEA0183
+      // messages
+      cluon::UDPReceiver connection{
+          address, port,
+          [&assembler](std::string &&d, std::string && /*from*/,
+                       std::chrono::system_clock::time_point &&tp) noexcept {
+            std::cout << d << std::endl;
+            assembler(d, std::move(tp));
+          }};
+
+      using namespace std::literals::chrono_literals;  // NOLINT
+      while (connection.isRunning()) {
+        std::this_thread::sleep_for(1s);
+      }
+
+    } else {
+      cluon::TCPConnection connection{
+          address, port,
+          [&assembler](std::string &&d,
+                       std::chrono::system_clock::time_point &&tp) noexcept {
+            assembler(d, std::move(tp));
+          },
+          [&argv]() {
+            std::cerr << "[" << argv[0] << "] Connection lost." << std::endl;
+            exit(1);
+          }};
+
+      // Just sleep as this microservice is data driven.
+      using namespace std::literals::chrono_literals;  // NOLINT
+      while (connection.isRunning()) {
+        std::this_thread::sleep_for(1s);
+      }
     }
   });