Manajemen pengiriman data menggunakan c++ queue

raingauge_queue_multithread.cpp

#include <thread>
#include <iostream>
#include <chrono>
#include <queue>
#include <mutex>
#include <vector>
#include <condition_variable>
#include <algorithm>
#include <ctime>
#include <stdlib.h>

using std::vector;
using std::thread;
using std::unique_lock;
using std::mutex;
using std::condition_variable;
using std::queue;
using std::chrono::seconds;
using std::chrono::system_clock;
using std::string;
using std::time_t;


struct Rainfall {
    string timeSampling;
    string sentAt;
    int counter;
    float humidity;
    float voltage;
    float temperature;
};

class WorkQueue {
    queue<Rainfall> work;

public:
    void push(Rainfall item) {
        work.push(item);
    }

    Rainfall pop() {
        Rainfall tmp = work.front();
        work.pop();
        return tmp;
    }

    queue<Rainfall> queue() {
        return work;
    }
};

string get_time_now() {
    time_t raw_time;
    struct tm * time_info;
    int buffer_size = 80;
    char buffer [buffer_size];

    time (&raw_time);
    time_info = localtime(&raw_time);
    strftime(buffer, buffer_size, "%Y-%m-%d %H:%M:%S", time_info);

    return buffer;
}


/**
 * Mensimulasikan keberhasilan/kegagalan
 * pengiriman data
 * @return bool
 */
bool simulasi_kirim_data() {
    srand (time(NULL));
    int random_number = rand() % 10;
    return (bool)(random_number % 2 == 0);
}

int main() {
    WorkQueue work_queue;

    auto producer = [&]() {
        while (true) {
            string now = get_time_now();
            srand (time(NULL));
            Rainfall rainfall = *new Rainfall{
                    now,
                    "",
                    std::rand() % 500,
                    static_cast<float>(std::rand() % 10),
                    static_cast<float>(std::rand() % 5),
                    static_cast<float>(std::rand() % 15),
            };

            /**
             * Tambah data ke antrian, setiap SLEEP_TIME
             */
            std::cout << "STORE . ";

            work_queue.push(rainfall);

            std::cout << now << " " << rainfall.counter;
            std::cout << " . END STORE" << std::endl;

            std::this_thread::sleep_for(seconds(1));  // SLEEP_TIME
        }
    };

    vector<thread> producers;
    producers.push_back(std::thread(producer));

    std::thread consumer([&]() {
        while (true) {

            /**
             * Ambil semua data yang ada diantrian saat ini
             */
            int queue_size = work_queue.queue().size();
            Rainfall *rainfalls = new Rainfall[queue_size];

            for (int i=0; i<queue_size; i++) {
                rainfalls[i] = work_queue.pop();  // inisiasi data
            }

            /**
             * [SIMULASI] Proses pengiriman data
             */
            std::cout << "SEND" << std::endl;

            for (int i=0; i<queue_size; i++) {
                // ganti tanggal pengiriman
                rainfalls[i].sentAt = get_time_now();

                std::cout << rainfalls[i].timeSampling << " " << rainfalls[i].counter << " ; ";
            }

            std::cout << std::endl;

            if (simulasi_kirim_data()) {
                std::cout << "VVV BERHASIL" << std::endl;
            } else {
                std::cout << "XXX GAGAL" << std::endl;

                /**
                 * Kembalikan data ke antrian
                 */
                for (int i=0; i<queue_size; i++) {
                    work_queue.push(rainfalls[i]);
                }
            }

            delete [] rainfalls;

            std::cout << std::endl << "END SEND" << std::endl;

            std::this_thread::sleep_for(seconds(5));
        }
    });

    std::for_each(producers.begin(), producers.end(), [](thread &p) {
        p.join();
    });

    consumer.join();
}