subscriber.h

Go to the documentation of this file.

// Copyright 2017 Space HAUC Command and Data Handling
// This file is part of octopOS which is released under AGPLv3.
// See file LICENSE.txt or go to <http://www.gnu.org/licenses/> for full
// license details.

#ifndef INCLUDE_SUBSCRIBER_H_
#define INCLUDE_SUBSCRIBER_H_

#include <pthread.h>

#include <unordered_map>
#include <sstream>
#include <vector>
#include <algorithm>
#include <iterator>
#include <string>
#include <queue>
#include <functional>
#include <tuple>

#include "tentacle.h"  // NOLINT

template< typename T >
class Optional {
 public:
    explicit Optional( T data ) : data_(data), isDefined_( true ) {}
    Optional() : isDefined_( false ) {}
    inline bool isDefined() { return isDefined_; }
    inline bool data() { return data_; }
 private:
     bool isDefined_;
     T data_;
};

typedef std::function<
    void(std::tuple<sem_id_t, shm_object*, generic_t*>&)> callback;

class subscriber_manager : protected tentacle {
 public:
    static void* wait_for_data(void* data);

 protected:

    static std::mutex topic_ids_lock,
           topic_memory_lock,
           registered_callbacks_lock;

    static bool register_cb(callback cb, std::string topic, uint size,
        subscriber_manager* sub);

    static std::unordered_map<std::string, std::tuple<sem_id_t, shm_object*,
        generic_t*>> topic_memory;

    static std::unordered_map<std::string, octopOS_id_t> topic_ids;

    static std::unordered_map<std::string, std::vector<callback>>
        registered_callbacks;

    explicit subscriber_manager(key_t shared_queue) : tentacle(shared_queue) {}
};

template <typename T>
class subscriber : public subscriber_manager {
 public:
    subscriber(std::string topic_name, key_t shared_queue)
        : subscriber_manager(shared_queue),
        topic(topic_name) {
        pthread_mutex_init(&data_queue_lock, NULL);
        pthread_cond_init(&data_queue_condition, NULL);

        if (!subscriber::register_cb(
            [this](std::tuple<sem_id_t, shm_object*, generic_t*> &data) {
            sem_id_t sem_id = std::get<0>(data);
            shm_object* rw = std::get<1>(data);

            if (p(sem_id, 3) < 0)
                exit(1);
            if (p(sem_id, 0) < 0)
                exit(1);
            rw->rw_array[0] += 1;
            if (rw->rw_array[0] == 1)
                if (p(sem_id, 2) < 0)
                    exit(1);
            v(sem_id, 0);
            v(sem_id, 3);

            pthread_mutex_lock(&data_queue_lock);
            data_queue.push(*reinterpret_cast<T*>(std::get<2>(data)));
            pthread_mutex_unlock(&data_queue_lock);

            if (p(sem_id, 0) < 0)
                exit(1);
            rw->rw_array[0] -= 1;
            if (rw->rw_array[0] == 0)
                if (v(sem_id, 2) < 0)
                    exit(1);
            v(sem_id, 0);
            pthread_cond_signal(&data_queue_condition);
        }, topic_name, sizeof(T), this))
        throw std::runtime_error("Failed to register topic");
    }

    bool data_available() {
        return !data_queue.empty();
    }

    T get_data() {
        T return_value;
        pthread_mutex_lock(&data_queue_lock);

        while (data_queue.empty())
             pthread_cond_wait(&data_queue_condition, &data_queue_lock);

        return_value = data_queue.front();
        data_queue.pop();

        pthread_mutex_unlock(&data_queue_lock);
        return return_value;
    }

    Optional<T> get_data_async() {
        pthread_mutex_lock(&data_queue_lock);
        if (data_queue.empty()) {
            pthread_mutex_unlock(&data_queue_lock);
            return Optional<T>();
        } else {
            Optional<T> return_value( data_queue.front() );
            data_queue.pop();
            pthread_mutex_unlock(&data_queue_lock);
            return return_value;
        }
    }

    ~subscriber() {
        pthread_mutex_destroy(&data_queue_lock);
        pthread_cond_destroy(&data_queue_condition);
    }

 private:
    std::string topic;

    pthread_mutex_t data_queue_lock;

    pthread_cond_t data_queue_condition;

    std::queue<T> data_queue;

    long id;                                                                      // NOLINT
};
#endif  // INCLUDE_SUBSCRIBER_H_