Line data    Source code

       1             : #include "GlobalState.hpp"
       2             : 
       3             : #include <exception>
       4             : #include <functional>
       5             : #include <stdexcept>
       6             : 
       7             : using namespace std;
       8             : 
       9           0 : GlobalState::ResourceException::ResourceException(
      10             :     const GlobalState::ResourceID &id_,
      11             :     const string                  &msg_
      12           0 : ):
      13             :     runtime_error(msg_),
      14           0 :     id(id_) {}
      15             : 
      16           0 : GlobalState::ResourceDoesNotExistException::ResourceDoesNotExistException(
      17             :     const GlobalState::ResourceID &id_
      18           0 : ):
      19           0 :     ResourceException(id_, "Resource " + id_ + " does not exist") {}
      20             : 
      21           0 : GlobalState::ResourceAlreadyExistsException::ResourceAlreadyExistsException(
      22             :     const GlobalState::ResourceID &id_
      23           0 : ):
      24           0 :     ResourceException(id_, "Resource " + id_ + " already exists") {}
      25             : 
      26           0 : utils::pipestream &GlobalState::Streams::create(const GlobalState::ResourceID &id) {
      27           0 :     lock_guard<mutex> lock(*this);
      28             : 
      29           0 :     auto [it, success] = GlobalState::streams->emplace(id, make_shared<utils::pipestream>());
      30           0 :     if(!success) {
      31           0 :         throw ResourceAlreadyExistsException(id);
      32             :     }
      33           0 :     return *it->second.get();
      34             : }
      35             : 
      36           0 : utils::pipestream &GlobalState::Streams::get(const GlobalState::ResourceID &id) {
      37           0 :     lock_guard<mutex> lock(*this);
      38             : 
      39           0 :     auto it = (*this)->find(id);
      40           0 :     if(it == streams->end()) {
      41           0 :         throw ResourceDoesNotExistException(id);
      42             :     }
      43           0 :     return *it->second.get();
      44             : }
      45             : 
      46           0 : void GlobalState::Streams::erase(const GlobalState::ResourceID &id) {
      47           0 :     lock_guard<mutex> lock(*this);
      48             : 
      49           0 :     (*this)->erase(id);
      50           0 : }
      51             : 
      52           0 : shared_future<GlobalState::TaskReturn> &GlobalState::Tasks::create(
      53             :     const GlobalState::ResourceID &id,
      54             :     const function<TaskReturn()>  &f
      55             : ) {
      56           0 :     lock_guard<mutex> lock(*this);
      57             : 
      58           0 :     auto [it, success] = GlobalState::tasks->emplace(
      59             :         id,
      60           0 :         make_shared<shared_future<GlobalState::TaskReturn>>()
      61           0 :     );
      62             : 
      63           0 :     if(!success) {
      64           0 :         throw ResourceAlreadyExistsException(id);
      65             :     }
      66             : 
      67           0 :     shared_future<TaskReturn> &future = *(it->second);
      68             : 
      69           0 :     future = shared_future<TaskReturn>(
      70           0 :         async(launch::async, [this, id, f]() -> TaskReturn {
      71           0 :             try {
      72           0 :                 TaskReturn ret = f();
      73             : 
      74           0 :                 lock_guard<mutex> taskLock(*this);
      75             : 
      76           0 :                 thread([this, id]() {
      77           0 :                     this->erase(id);
      78           0 :                 }).detach();
      79             : 
      80           0 :                 return ret;
      81           0 :             } catch(const exception &e) {
      82           0 :                 return {500, "what(): "s + e.what()};
      83             :             }
      84             :         })
      85           0 :     );
      86             : 
      87           0 :     return future;
      88             : }
      89             : 
      90           0 : shared_future<GlobalState::TaskReturn> &GlobalState::Tasks::get(
      91             :     const GlobalState::ResourceID &id
      92             : ) {
      93           0 :     lock_guard<mutex> lock(*this);
      94             : 
      95           0 :     auto it = (*this)->find(id);
      96           0 :     if(it == tasks->end()) {
      97           0 :         throw ResourceDoesNotExistException(id);
      98             :     }
      99           0 :     return *it->second.get();
     100             : }
     101             : 
     102           0 : void GlobalState::Tasks::erase(const GlobalState::ResourceID &id) {
     103           0 :     lock_guard<mutex> lock(*this);
     104             : 
     105           0 :     auto it = (*this)->find(id);
     106             : 
     107           0 :     const auto &[_, future] = *it;
     108             : 
     109           0 :     future->wait();
     110             : 
     111           0 :     (*this)->erase(it);
     112           0 : }
     113             : 
     114             : GlobalState::Streams GlobalState::streams;
     115             : 
     116             : GlobalState::Tasks GlobalState::tasks;