IoT Data Collection System. More...

Collaboration diagram for scenarios::IoTDataCollectionSystem:

Classes
struct	SensorReading

Public Member Functions
void	simulate_iot_scenario ()

Private Member Functions
void	send_iot_batch (const std::vector< SensorReading > &batch)

Private Attributes
std::atomic< int >	readings_collected_ {0}

std::atomic< int >	batches_sent_ {0}

Detailed Description

IoT Data Collection System.

Simulates collecting sensor data from multiple IoT devices, aggregating them, and sending to a central processing system.

Examples: real_world_scenarios.cpp.

Definition at line 52 of file real_world_scenarios.cpp.

Member Function Documentation

◆ send_iot_batch()

void scenarios::IoTDataCollectionSystem::send_iot_batch ( const std::vector< SensorReading > & batch )

inlineprivate

Examples: real_world_scenarios.cpp.

Definition at line 154 of file real_world_scenarios.cpp.

                                                               {
        auto container = std::make_shared<value_container>();
        container->set_source("iot_aggregator", "batch_processor");
        container->set_target("iot_analytics_service", "data_processor");
        container->set_message_type("sensor_data_batch");
 
        container->set("batch_size", static_cast<int32_t>(batch.size()));
        container->set("batch_timestamp", static_cast<int64_t>(
            std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::system_clock::now().time_since_epoch()).count()));
 
        for (size_t i = 0; i < batch.size(); ++i) {
            const auto& reading = batch[i];
            std::string prefix = "reading_" + std::to_string(i) + "_";
 
            container->set(prefix + "device_id", reading.device_id);
            container->set(prefix + "sensor_type", reading.sensor_type);
            container->set(prefix + "value", reading.value);
            container->set(prefix + "timestamp", static_cast<int64_t>(
                std::chrono::duration_cast<std::chrono::milliseconds>(
                    reading.timestamp.time_since_epoch()).count()));
        }
 
        std::string serialized = container->serialize_string(value_container::serialization_format::binary).value();
        std::cout << "  Sent IoT batch: " << batch.size() << " readings, "
                  << serialized.size() << " bytes" << std::endl;
    }

Referenced by simulate_iot_scenario().

Here is the caller graph for this function:

◆ simulate_iot_scenario()

void scenarios::IoTDataCollectionSystem::simulate_iot_scenario ( )

inline

Examples: real_world_scenarios.cpp.

Definition at line 65 of file real_world_scenarios.cpp.

                                 {
        std::cout << "\n=== IoT Data Collection Scenario ===" << std::endl;
 
        const int num_devices = 5;
        const int readings_per_device = 20;
        const int batch_size = 10;
 
        std::vector<std::string> device_types = {"temperature", "humidity", "pressure"};
        std::vector<std::thread> device_threads;
 
        std::queue<SensorReading> sensor_queue;
        std::mutex queue_mutex;
        std::condition_variable queue_cv;
        std::atomic<bool> collection_active{true};
 
        // Data aggregator thread
        std::thread aggregator_thread([&]() {
            std::vector<SensorReading> batch;
            batch.reserve(batch_size);
 
            while (collection_active || !sensor_queue.empty()) {
                std::unique_lock<std::mutex> lock(queue_mutex);
                queue_cv.wait(lock, [&]() { return !sensor_queue.empty() || !collection_active; });
 
                while (!sensor_queue.empty() && batch.size() < static_cast<size_t>(batch_size)) {
                    batch.push_back(sensor_queue.front());
                    sensor_queue.pop();
                }
 
                if (!batch.empty()) {
                    lock.unlock();
                    send_iot_batch(batch);
                    batch.clear();
                    batches_sent_++;
                }
            }
        });
 
        // Device simulation threads
        std::random_device rd;
        for (int device_id = 0; device_id < num_devices; ++device_id) {
            device_threads.emplace_back([&, device_id]() {
                std::mt19937 gen(rd());
                std::uniform_real_distribution<> temp_dist(18.0, 35.0);
                std::uniform_real_distribution<> humidity_dist(30.0, 80.0);
                std::uniform_real_distribution<> pressure_dist(990.0, 1030.0);
 
                for (int reading = 0; reading < readings_per_device; ++reading) {
                    for (const auto& sensor_type : device_types) {
                        SensorReading sensor_reading;
                        sensor_reading.device_id = "device_" + std::to_string(device_id);
                        sensor_reading.sensor_type = sensor_type;
                        sensor_reading.timestamp = std::chrono::system_clock::now();
 
                        if (sensor_type == "temperature") {
                            sensor_reading.value = temp_dist(gen);
                        } else if (sensor_type == "humidity") {
                            sensor_reading.value = humidity_dist(gen);
                        } else {
                            sensor_reading.value = pressure_dist(gen);
                        }
 
                        {
                            std::lock_guard<std::mutex> lock(queue_mutex);
                            sensor_queue.push(sensor_reading);
                        }
                        queue_cv.notify_one();
                        readings_collected_++;
 
                        std::this_thread::sleep_for(std::chrono::milliseconds(5));
                    }
                }
            });
        }
 
        for (auto& thread : device_threads) {
            thread.join();
        }
 
        collection_active = false;
        queue_cv.notify_all();
        aggregator_thread.join();
 
        std::cout << "IoT simulation completed:" << std::endl;
        std::cout << "  Readings collected: " << readings_collected_.load() << std::endl;
        std::cout << "  Batches sent: " << batches_sent_.load() << std::endl;
    }