Обратные вызовы в C++ - [45]

Листинг 93. Класс для выполнения асинхронных вызовов (CommandQueue.h)

class CommandQueue
{
public:
  void start();  // (1)
  void stop();   // (2)
  void addCommand(SensorNumber number, SensorPointer pointer, SensorValueCallback callback);  // (3)

private:
  struct Command  // (4)
  {
    SensorNumber number;
    SensorPointer pointer;
    SensorValueCallback callback;
  };

  std::queue commandQueue_ ;    // (5)
  std::condition_variable conditional_;  // (6)
  std::mutex mutex_;                     // (7)
  std::thread queueThread_;              // (8)
  bool exit_;                            // (9)

  void readCommand();  // (10)
};

В строке 4 объявлена структура, в которой будут храниться данные для выполнения вызова: номер датчика, указатель на класс датчика и объект вызова. В строке 5 объявлен контейнер, который будет хранить указанные структуры. В строках 6 и 7 объявлены переменные для синхронизации операций записи/чтения очереди, в строке 8 объявлен класс для запуска потока обработки очереди, в строке 9 объявлен индикатор для завершения работы потока.

В строке 1 объявлен метод, который запускает поток обработки очереди, в строке 2 объявлен метод для остановки этого потока. Метод, объявленный в строке 3, добавляет переданные данные в очередь путем создания экземпляра структуры 4 и размещения ее в контейнере 5.

Обработка очереди реализована в методе, объявленном в строке 10. Поток обработки очереди вызывает этот метод, который, в свою очередь, ожидает поступления записей и обрабатывает их. Реализация приведена в Листинг 95.

Листинг 94. Обработка очереди запросов (CommandQueue.cpp)

void CommandQueue::readCommand()
{
  while (!exit_)  // (1)
  {
    std::unique_lock lock(mutex_);  // (2)

    conditional_.wait(lock, [this]() {return commandQueue_.size() > 0 || exit_ == true; });  // (3)

    while (commandQueue_.size() > 0 && exit_ == false)      // (4)
    {
      Command cmd = commandQueue_.front();                // (5)
      commandQueue_.pop();                                // (6)
      lock.unlock();                                      // (7)
      cmd.callback(cmd.number, cmd.pointer->getValue());  // (8)
      lock.lock();  // (9)
    }
  }
}

Пока не установлен индикатор завершения (устанавливается в методе stop), выполняется цикл 1. Вначале блокируется мьютекс 2 (это необходимо для корректной работы условной переменной), затем осуществляется ожидание условной переменной 3. Когда метод addCommand сформировал новую запись и добавил ее в контейнер, он инициирует срабатывание условной переменной, и поток выполнения переходит к циклу 4 (мьютекс при этом оказывается заблокирован). Этот цикл работает, пока очередь не опустеет либо будет установлен индикатор выхода.

В строке 5 из контейнера извлекается очередная запись, в строке 6 эта запись удаляется из контейнера. В строке 7 снимается блокировка мьютекса, что позволяет добавлять в контейнер новые записи, пока идет обработка очередной команды. В строке 8 осуществляется обратный вызов, в строке 9 мьютекс блокируется вновь, и далее повторяется цикл 4.

6.2.7. Наблюдатель

Объявление класса наблюдателя приведено в Листинг 95.

Листинг 95. Наблюдатель – класс для отслеживания пороговых значений (Observer.h)

class Observer
{
public:
  void start();  // (1)
  void stop();   // (2)
  void addAlert(SensorNumber number, SensorPointer pointer, SensorAlertCallback callback, SensorValue alertValue, AlertRule alertRule, CheckAlertTimeout checkTimeoutSeс);  // (3)
  void deleteAlert(SensorNumber number);  // (4)
private:

  struct Alert  // (5)
  {
    Alert() {}
    Alert(SensorAlertCallback callback, SensorValue alertValue, AlertRule alertRule, SensorPointer sensor, CheckAlertTimeout checkTimeout):
    callback(callback), alertValue(alertValue), alertRule(alertRule), sensor(sensor), checkTimeout(checkTimeout), currentTimeout(0)
    {
    }

    SensorAlertCallback callback;
    SensorValue alertValue;
    AlertRule alertRule;
    SensorPointer sensor;
    CheckAlertTimeout checkTimeout;
    CheckAlertTimeout currentTimeout;
  };

  std::map containerAlert;  // (6)
  std::thread pollThread_;                       // (7)
  bool exit_;                                    // (8)
  std::mutex mutex_;                             // (9)

  void poll();                                   // (10)
};

В строке 1 объявлен метод для запуска процесса отслеживания пороговых значений, в строке 2 – метод для останова. Метод в строке 3 добавляет датчик для отслеживания, метод 4 – удаляет.

В строке 5 объявлена структура, в которой хранятся данные, необходимые для отслеживания показаний датчика. В строке 6 объявлен контейнер для хранения указанных структур; метод addAlert добавляет запись в контейнер, метод deleteAlert удаляет ее. В строке 7 объявлен класс для запуска потока для отслеживания, в строке 8 объявлен индикатор выхода, в строке 9 объявлен мьютекс для синхронизации.

Отслеживание показаний реализовано в методе, объявленном в строке 10. Поток отслеживания вызывает этот метод, который циклически опрашивает назначенные датчики и в случае превышения пороговых значений осуществляет обратный вызов. Реализация приведена в Листинг 96.