comm_drv_signalk_net.cpp

Go to the documentation of this file.

/***************************************************************************
 *   Copyright (C) 2022 by David Register                                  *
 *   Copyright (C) 2022 Alec Leamas                                        *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, see <https://www.gnu.org/licenses/>. *
 **************************************************************************/
 
#include <chrono>
#include <mutex>  // std::mutex
 
#include <wx/socket.h>
 
#include "rapidjson/document.h"
#include "ixwebsocket/IXNetSystem.h"
#include "ixwebsocket/IXSocketTLSOptions.h"
#include "ixwebsocket/IXWebSocket.h"
#include "observable.h"
 
#include "model/comm_drv_signalk_net.h"
#include "model/comm_navmsg_bus.h"
#include "model/geodesic.h"
#include "model/logger.h"
#include "model/sys_events.h"
#include "model/thread_ctrl.h"
#include "wxServDisc.h"
 
using namespace std::literals::chrono_literals;
 
constexpr int kTimerSocket = 9006;
constexpr int kSignalkSocketId = 5011;
constexpr int kDogTimeoutReconnectSeconds = 10;
 
constexpr double kMsToKnotFactor = 1.9438444924406;
 
class CommDriverSignalKNet::IoThread : public ThreadCtrl {
public:
  IoThread(const std::string& iface, const wxIPV4address& address,
           wxEvtHandler* consumer, const std::string& token);
 
  ~IoThread() override = default;
 
  void Run();
 
  DriverStats GetStats() const;
 
private:
  wxIPV4address m_address;
  wxEvtHandler* m_consumer;
  const std::string m_iface;
  std::string m_token;
  ix::WebSocket m_ws;
  ObsListener m_resume_listener;
  DriverStats m_driver_stats;
  mutable std::mutex m_stats_mutex;
};
 
// i. e. wxDEFINE_EVENT(), avoiding the evil macro.
static const wxEventTypeTag<CommDriverSignalKNet::InputEvt> SignalkEvtType(
    wxNewEventType());
 
static wxIPV4address ParamsIpAddress(const ConnectionParams& params) {
  wxIPV4address addr;
  addr.Hostname(params.NetworkAddress);
  addr.Service(params.NetworkPort);
  return addr;
}
 
class CommDriverSignalKNet::InputEvt : public wxEvent {
public:
  explicit InputEvt(std::string payload)
      : wxEvent(0, SignalkEvtType), m_payload(std::move(payload)) {};
 
  std::string GetPayload() const { return m_payload; }
 
  // required for sending with wxPostEvent()
  wxEvent* Clone() const override { return new InputEvt(m_payload); };
 
private:
  const std::string m_payload;
};
 
//========================================================================
//      IoThread implementation
//
CommDriverSignalKNet::IoThread::IoThread(const std::string& iface,
                                         const wxIPV4address& address,
                                         wxEvtHandler* consumer,
                                         const std::string& token)
    : m_address(address), m_consumer(consumer), m_iface(iface), m_token(token) {
  m_resume_listener.Init(SystemEvents::GetInstance().evt_resume,
                         [&](ObservedEvt& ev) {
                           m_ws.stop();
                           m_ws.start();
                           wxLogDebug("WebSocketThread: restarted");
                         });
}
 
void CommDriverSignalKNet::IoThread::Run() {
  using namespace std::chrono_literals;
 
  {
    std::lock_guard lock(m_stats_mutex);
    m_driver_stats.driver_bus = NavAddr::Bus::Signalk;
    m_driver_stats.driver_iface = m_iface;
    m_driver_stats.available = false;
  }
 
  // Craft the address strings
  wxString host = m_address.IPAddress();
  int port = m_address.Service();
  std::stringstream wsAddress;
  wsAddress << "ws://" << host << ":" << port
            << "/signalk/v1/stream?subscribe=all&sendCachedValues=false";
  if (!m_token.empty()) wsAddress << "&token=" << m_token;
  std::stringstream wssAddress;
  wssAddress << "wss://" << host << ":" << port
             << "/signalk/v1/stream?subscribe=all&sendCachedValues=false";
  if (!m_token.empty()) wssAddress << "&token=" << m_token;
 
  m_ws.setUrl(wssAddress.str());
  ix::SocketTLSOptions opt;
  opt.disable_hostname_validation = true;
  opt.caFile = "NONE";
  m_ws.setTLSOptions(opt);
  m_ws.setPingInterval(30);
 
  auto message_callback = [&](const ix::WebSocketMessagePtr& msg) {
    if (msg->type == ix::WebSocketMessageType::Message) {
      m_consumer->QueueEvent(new InputEvt(msg->str));
      m_driver_stats.rx_count++;
    } else if (msg->type == ix::WebSocketMessageType::Open) {
      wxLogDebug("websocket: Connection to %s established",
                 m_ws.getUrl().c_str());
      std::lock_guard lock(m_stats_mutex);
      m_driver_stats.available = true;
    } else if (msg->type == ix::WebSocketMessageType::Close) {
      wxLogDebug("websocket: Connection disconnected");
      std::lock_guard lock(m_stats_mutex);
      m_driver_stats.available = false;
    } else if (msg->type == ix::WebSocketMessageType::Error) {
      std::lock_guard lock(m_stats_mutex);
      m_driver_stats.error_count++;
      wxLogDebug("websocket: error: %s", msg->errorInfo.reason.c_str());
      m_ws.getUrl() == wsAddress.str() ? m_ws.setUrl(wssAddress.str())
                                       : m_ws.setUrl(wsAddress.str());
    }
  };
  m_ws.setOnMessageCallback(message_callback);
 
  m_ws.start();
  while (KeepGoing()) {
    std::this_thread::sleep_for(100ms);
  }
  m_ws.stop();
  SignalExit();
  std::lock_guard lock(m_stats_mutex);
  m_driver_stats.available = false;
}
 
//========================================================================
//    CommDriverSignalKNet implementation
//
DriverStats CommDriverSignalKNet::IoThread::GetStats() const {
  std::lock_guard lock(m_stats_mutex);
  return m_driver_stats;
}
 
CommDriverSignalKNet::CommDriverSignalKNet(const ConnectionParams* params,
                                           DriverListener& listener)
    : CommDriverSignalK(params->GetStrippedDSPort()),
      m_params(*params),
      m_listener(listener),
      m_dog_value(kDogTimeoutSeconds),
      m_io_thread(std::make_unique<IoThread>(params->GetStrippedDSPort(),
                                             ParamsIpAddress(*params), this,
                                             params->AuthToken.ToStdString())),
      m_stats_timer(*this, 2s) {
  // Prepare the wxEventHandler to accept events from the actual hardware thread
  Bind(SignalkEvtType, &CommDriverSignalKNet::HandleSkSentence, this);
 
  m_socketread_watchdog_timer.SetOwner(this, kTimerSocket);
 
  // Dummy Driver Stats, may be polled before worker thread is active
  m_driver_stats.driver_bus = NavAddr::Bus::Signalk;
  m_driver_stats.driver_iface = m_params.GetStrippedDSPort();
  m_driver_stats.available = false;
 
  Open();
}
 
CommDriverSignalKNet::~CommDriverSignalKNet() { Close(); }
 
DriverStats CommDriverSignalKNet::GetDriverStats() const {
  if (m_std_thread.joinable())
    return m_io_thread->GetStats();
  else
    return m_driver_stats;
}
 
void CommDriverSignalKNet::Open() {
  wxString discoveredIP;
#if 0
  int discoveredPort;
#endif
 
  // if (m_useWebSocket)
  {
    std::string service_ident =
        std::string("_signalk-ws._tcp.local.");  // Works for node.js server
    OpenWebSocket();
  }
}
void CommDriverSignalKNet::Close() { CloseWebSocket(); }
 
bool CommDriverSignalKNet::DiscoverSkServer(const std::string& service_ident,
                                            wxString& ip, int& port, int tSec) {
  auto servscan = std::make_unique<wxServDisc>(
      nullptr, wxString(service_ident.c_str()), QTYPE_PTR);
  for (int i = 0; i < 10; i++) {
    if (servscan->getResultCount()) {
      auto result = servscan->getResults().at(0);
      auto namescan =
          std::make_unique<wxServDisc>(nullptr, result.name, QTYPE_SRV);
      for (int j = 0; j < 10; j++) {
        if (namescan->getResultCount()) {
          auto namescan_result = namescan->getResults().at(0);
          port = namescan_result.port;
          auto addrscan = std::make_unique<wxServDisc>(
              nullptr, namescan_result.name, QTYPE_A);
          for (int k = 0; k < 10; k++) {
            if (addrscan->getResultCount()) {
              auto addrscan_result = addrscan->getResults().at(0);
              ip = addrscan_result.ip;
              return true;
            } else {
              wxYield();
              wxMilliSleep(1000 * tSec / 10);
            }
          }
          return false;
        } else {
          wxYield();
          wxMilliSleep(1000 * tSec / 10);
        }
      }
      return false;
    } else {
      wxYield();
      wxMilliSleep(1000 * tSec / 10);
    }
  }
  return false;
}
 
void CommDriverSignalKNet::OpenWebSocket() {
  wxLogMessage("Opening Signal K WebSocket client: %s",
               m_params.GetDSPort().c_str());
  m_std_thread = std::thread([&] { m_io_thread->Run(); });
  if (!m_std_thread.joinable()) {
    wxLogError("Can't create WebSocketThread!");
    return;
  }
  ResetWatchdog();
  m_socketread_watchdog_timer.Start(1000, wxTIMER_ONE_SHOT);
}
 
void CommDriverSignalKNet::CloseWebSocket() {
  if (m_std_thread.joinable()) {
    if (m_io_thread->IsRunning()) {
      wxLogMessage("Stopping Secondary SignalK Thread");
      m_stats_timer.Stop();
      m_io_thread->RequestStop();
      std::chrono::milliseconds stop_delay;
      bool stop_ok = m_io_thread->WaitUntilStopped(10s, stop_delay);
      if (stop_ok)
        MESSAGE_LOG << "Stopped in" << stop_delay.count() << " msec.";
      else
        WARNING_LOG << "Not stopped after 10 sec.";
    }
    wxMilliSleep(100);
    m_std_thread.join();
  } else {
    WARNING_LOG << "Thread unexpectedly died";
  }
}
 
void CommDriverSignalKNet::HandleSkSentence(const InputEvt& event) {
  rapidjson::Document root;
 
  std::string msg = event.GetPayload();
  root.Parse(msg);
  if (root.HasParseError()) {
    wxLogMessage(
        "SignalKDataStream ERROR: the JSON document is not well-formed: %d",
        root.GetParseError());
    return;
  }
  if (!root.IsObject()) {
    wxLogMessage("SignalKDataStream ERROR: Message is not a JSON Object: %s",
                 msg.c_str());
    return;
  }
 
  // Decode just enough of string to extract some identifiers
  // such as the sK version, "self" context, and target context
  if (root.HasMember("version")) {
    wxString vers = "Connected to Signal K server version: ";
    vers << (root["version"].GetString());
    wxLogMessage(vers);
  }
  if (root.HasMember("self")) {
    if (strncmp(root["self"].GetString(), "vessels.", 8) == 0)
      m_self = (root["self"].GetString());  // for java server, and OpenPlotter
                                            // node.js server 1.20
    else
      m_self = std::string("vessels.")
                   .append(root["self"].GetString());  // for Node.js server
  }
  if (root.HasMember("context") && root["context"].IsString()) {
    m_context = root["context"].GetString();
  }
 
  // Notify all listeners
  auto pos = iface.find(':');
  std::string comm_interface;
  if (pos != std::string::npos) comm_interface = iface.substr(pos + 1);
  auto navmsg = std::make_shared<const SignalkMsg>(m_self, m_context, msg,
                                                   comm_interface);
  m_listener.Notify(std::move(navmsg));
}
 
void CommDriverSignalKNet::initIXNetSystem() { ix::initNetSystem(); };
 
void CommDriverSignalKNet::uninitIXNetSystem() { ix::uninitNetSystem(); };