http_shared.h

Go to the documentation of this file.

/*
 * This file is part of OpenTTD.
 * OpenTTD is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, version 2.
 * OpenTTD is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with OpenTTD. If not, see <https://www.gnu.org/licenses/old-licenses/gpl-2.0>.
 */

 
#ifndef NETWORK_CORE_HTTP_SHARED_H
#define NETWORK_CORE_HTTP_SHARED_H
 
#include "http.h"
 
#include <condition_variable>
#include <mutex>
#include <vector>

class HTTPThreadSafeCallback {
private:
  class Callback {
  public:
    Callback(std::unique_ptr<char[]> data, size_t length) : data(std::move(data)), length(length), failure(false) {}
    Callback() : data(nullptr), length(0), failure(true) {}
 
    std::unique_ptr<char[]> data;
    size_t length;
    bool failure;
  };
 
public:
  void OnFailure()
  {
    std::lock_guard<std::mutex> lock(this->mutex);
    this->queue.emplace_back();
  }

  void OnReceiveData(std::unique_ptr<char[]> data, size_t length)
  {
    std::lock_guard<std::mutex> lock(this->mutex);
    this->queue.emplace_back(std::move(data), length);
  }

  void HandleQueue()
  {
    this->cancelled = callback->IsCancelled();
 
    std::lock_guard<std::mutex> lock(this->mutex);
 
    for (auto &item : this->queue) {
      if (item.failure) {
        this->callback->OnFailure();
      } else {
        this->callback->OnReceiveData(std::move(item.data), item.length);
      }
    }
 
    this->queue.clear();
    this->queue_cv.notify_all();
  }

  template <typename T>
  void WaitTillEmptyOrCondition(T condition)
  {
    std::unique_lock<std::mutex> lock(this->mutex);
 
    while (!(queue.empty() || condition())) {
      this->queue_cv.wait(lock);
    }
  }

  bool IsQueueEmpty()
  {
    std::lock_guard<std::mutex> lock(this->mutex);
    return this->queue.empty();
  }
 
  HTTPThreadSafeCallback(HTTPCallback *callback) : callback(callback) {}
 
  ~HTTPThreadSafeCallback()
  {
    std::lock_guard<std::mutex> lock(this->mutex);
 
    /* Clear the list and notify explicitly. */
    queue.clear();
    queue_cv.notify_all();
  }
 
  std::atomic<bool> cancelled = false;
 
private:
  HTTPCallback *callback; 
  std::mutex mutex; 
  std::vector<Callback> queue; 
  std::condition_variable queue_cv; 
};
 
#endif /* NETWORK_CORE_HTTP_SHARED_H */