OCLib/InProcClientWrapper.cpp

   1 //******************************************************************
   2 //
   3 // Copyright 2014 Intel Corporation All Rights Reserved.
   4 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
   5
   6
   7 #include "OCResource.h"
   8 #include "InProcClientWrapper.h"
   9
  10 #include "ocstack.h"
  11
  12 using namespace std;
  13
  14
  15 namespace OC
  16 {
  17     InProcClientWrapper::InProcClientWrapper(PlatformConfig cfg)
  18     {
  19         OCStackResult result = OCInit(cfg.ipAddress.c_str(), cfg.port, OC_CLIENT);
  20
  21         if(OC_STACK_OK != result)
  22         {
  23             throw InitializeException("Error Initializing Stack", result);
  24         }
  25
  26         m_threadRun = true;
  27         m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
  28     }
  29
  30     InProcClientWrapper::~InProcClientWrapper()
  31     {
  32         if(m_listeningThread.joinable())
  33         {
  34             m_threadRun = false;
  35             m_listeningThread.join();
  36         }
  37
  38         OCStop();
  39     }
  40
  41     void InProcClientWrapper::listeningFunc()
  42     {
  43         while(m_threadRun)
  44         {
  45                         OCStackResult result;
  46                         {
  47                                 std::lock_guard<std::mutex> lock(m_csdkLock);
  48                                 result = OCProcess();
  49                         }
  50
  51             if(result != OC_STACK_OK)
  52             {
  53                 // TODO: @Erich do something with result if failed?
  54             }
  55
  56             std::this_thread::yield();
  57             // To minimize CPU utilization we may wish to do this with sleep
  58                         //std::this_thread::sleep_for(std::chrono::milliseconds(1));
  59         }
  60     }
  61
  62
  63
  64     std::string convertOCAddrToString(OCDevAddr* addr)
  65     {
  66         if(addr->size != 4) { return "NOT SUPPORTED ADDR;";}
  67
  68         if(addr->size == 4) // IPV4
  69         {
  70                 std::ostringstream address;
  71                 address<<"coap://"<<addr->addr[0]<<"."<<addr->addr[1]<<"."<<addr->addr[2]<<"."<<addr->addr[3]<<"/";
  72                 return address.str();
  73         }
  74
  75         // TODO: @Erich Convert the device address to a valid string!
  76         return "";
  77     }
  78
  79
  80     OCStackApplicationResult listenCallback(void* ctx, OCClientResponse* clientResponse)
  81     {
  82         auto &callback =*(std::function <void(OCResource::Ptr)>*)ctx;
  83         std::stringstream requestStream;
  84         requestStream << clientResponse->resJSONPayload;
  85
  86
  87         boost::property_tree::ptree root;
  88         boost::property_tree::read_json(requestStream, root);
  89
  90         boost::property_tree::ptree payload = root.get_child("oc.payload", boost::property_tree::ptree());
  91
  92
  93         for(auto payloadItr : payload)
  94         {
  95                 try
  96                 {
  97                         std::string host = convertOCAddrToString(clientResponse->addr);
  98                         OCResource::Ptr resource = std::make_shared<OCResource>(host, payloadItr.second);
  99
 100                         // Note: the call to detach allows the underlying thread to continue until completion
 101                         //  and allows us to destroy the exec object.
 102                         //  This is apparently NOT a memory leak, as the thread will apparently take care of itself.
 103                         //  Additionally, the only parameter here is
 104                         //  a shared ptr, so OCResource will be disposed of properly upon completion of the callback handler.
 105                         std::thread exec(callback,resource);
 106                         exec.detach();
 107                 }
 108                 catch(ResourceInitException)
 109                 {
 110                         // TODO: Do we want to handle this somehow?  Perhaps we need to log this?
 111                 }
 112
 113         }
 114         delete clientResponse;
 115
 116         return OC_STACK_KEEP_TRANSACTION;
 117     }
 118
 119     int InProcClientWrapper::ListenForResource(const std::string& serviceUrl, const std::string& resourceType, std::function<void (OCResource::Ptr)>& callback)
 120     {
 121         OCStackResult result;
 122
 123         OCCallbackData* cbdata = new OCCallbackData();
 124         cbdata->context = (void*)(&callback);
 125         cbdata->cb = &listenCallback;
 126         {
 127                 std::lock_guard<std::mutex> lock(m_csdkLock);
 128
 129                 result = OCDoResource(OC_REST_GET, resourceType.c_str(), nullptr, nullptr, OC_NON_CONFIRMABLE, cbdata);
 130
 131         }
 132                 return result;
 133     }
 134    }