Client.hpp

Go to the documentation of this file.

#ifndef CLIENT_HPP
#define CLIENT_HPP

#include <asio.hpp>
#include <condition_variable>
#include <functional>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include "ByteVector.hpp"
#include "FirmwareVariants.hpp"
#include "Message.hpp"
#include "Subscription.hpp"

namespace msp {
namespace client {

typedef asio::buffers_iterator<asio::streambuf::const_buffers_type> iterator;

enum LoggingLevel { SILENT, WARNING, INFO, DEBUG };

enum MessageStatus {
    OK,       // no errors
    FAIL_ID,  // message ID is unknown
    FAIL_CRC  // wrong CRC
};

struct ReceivedMessage {
    msp::ID id;
    ByteVector payload;
    MessageStatus status;
};

class Client {
public:
    Client();

    ~Client();

    void setLoggingLevel(const LoggingLevel& level);

    bool setVersion(const int& ver);

    int getVersion() const;

    void setVariant(const FirmwareVariant& v);

    FirmwareVariant getVariant() const;

    bool start(const std::string& device, const size_t baudrate = 115200);

    bool stop();

    bool isConnected() const;

    bool sendMessage(msp::Message& message, const double& timeout = 0);

    bool sendMessageNoWait(const msp::Message& message);

    template <typename T, typename C,
              class = typename std::enable_if<
                  std::is_base_of<msp::Message, T>::value>::type>
    std::shared_ptr<SubscriptionBase> subscribe(void (C::*callback)(const T&),
                                                C* context, const double& tp) {
        return subscribe<T>(std::bind(callback, context, std::placeholders::_1),
                            tp);
    }

    template <typename T, class = typename std::enable_if<
                              std::is_base_of<msp::Message, T>::value>::type>
    std::shared_ptr<SubscriptionBase> subscribe(
        const std::function<void(const T&)>& recv_callback, const double& tp) {
        // validate the period
        if(!(tp >= 0.0)) throw std::runtime_error("Period must be positive!");

        // get the id of the message in question
        const msp::ID id = T(fw_variant).id();
        if(log_level_ >= INFO)
            std::cout << "SUBSCRIBING TO " << id << std::endl;

        // generate the callback for sending messages
        std::function<bool(const Message&)> send_callback =
            std::bind(&Client::sendMessageNoWait, this, std::placeholders::_1);

        // create a shared pointer to a new Subscription and set all properties
        auto subscription = std::make_shared<Subscription<T>>(
            recv_callback, send_callback, std::make_unique<T>(fw_variant), tp);

        // gonna modify the subscription map, so lock the mutex
        std::lock_guard<std::mutex> lock(mutex_subscriptions);

        // move the new subscription into the subscription map
        subscriptions.emplace(id, std::move(subscription));
        return subscriptions[id];
    }

    bool hasSubscription(const msp::ID& id) const {
        return (subscriptions.count(id) == 1);
    }

    std::shared_ptr<SubscriptionBase> getSubscription(const msp::ID& id) {
        return subscriptions.at(id);
    }

    void processOneMessage(const asio::error_code& ec,
                           const std::size_t& bytes_transferred);

    bool sendData(const msp::ID id, const ByteVector& data = ByteVector(0));

    bool sendData(const msp::ID id, const ByteVectorUptr&& data) {
        if(!data) return sendData(id);
        return sendData(id, *data);
    }

protected:
    bool connectPort(const std::string& device, const size_t baudrate = 115200);

    bool disconnectPort();

    bool startReadThread();

    bool stopReadThread();

    bool startSubscriptions();

    bool stopSubscriptions();

    uint8_t extractChar();

    std::pair<iterator, bool> messageReady(iterator begin, iterator end) const;

    ReceivedMessage processOneMessageV1();

    ReceivedMessage processOneMessageV2();

    ByteVector packMessageV1(const msp::ID id,
                             const ByteVector& data = ByteVector(0)) const;

    uint8_t crcV1(const uint8_t id, const ByteVector& data) const;

    ByteVector packMessageV2(const msp::ID id,
                             const ByteVector& data = ByteVector(0)) const;

    uint8_t crcV2(uint8_t crc, const ByteVector& data) const;

    uint8_t crcV2(uint8_t crc, const uint8_t& b) const;

protected:
    asio::io_service io;     
    asio::serial_port port;  
    asio::streambuf buffer;

    // read thread management
    std::thread thread;
    std::atomic_flag running_ = ATOMIC_FLAG_INIT;

    // thread safety and synchronization
    std::condition_variable cv_response;
    std::mutex cv_response_mtx;
    std::mutex mutex_response;
    std::mutex mutex_buffer;
    std::mutex mutex_send;

    // holder for received data
    std::unique_ptr<ReceivedMessage> request_received;

    // subscription management
    std::mutex mutex_subscriptions;
    std::map<msp::ID, std::shared_ptr<SubscriptionBase>> subscriptions;

    // debugging
    LoggingLevel log_level_;

    // reference values
    int msp_ver_;
    FirmwareVariant fw_variant;
};

}  // namespace client
}  // namespace msp

#endif  // CLIENT_HPP