src/InProcClientWrapper.cpp

   1 //******************************************************************
   2 //
   3 // Copyright 2014 Intel Mobile Communications GmbH All Rights Reserved.
   4 //
   5 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
   6 //
   7 // Licensed under the Apache License, Version 2.0 (the "License");
   8 // you may not use this file except in compliance with the License.
   9 // You may obtain a copy of the License at
  10 //
  11 //      http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 // Unless required by applicable law or agreed to in writing, software
  14 // distributed under the License is distributed on an "AS IS" BASIS,
  15 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 // See the License for the specific language governing permissions and
  17 // limitations under the License.
  18 //
  19 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
  20
  21 #include <new>
  22
  23 #include "InProcClientWrapper.h"
  24 #include "ocstack.h"
  25
  26 #include "OCPlatform.h"
  27 #include "OCResource.h"
  28
  29 using namespace std;
  30
  31 namespace OC
  32 {
  33     InProcClientWrapper::InProcClientWrapper(OC::OCPlatform& owner,
  34         std::weak_ptr<std::recursive_mutex> csdkLock, PlatformConfig cfg)
  35             : IClientWrapper(owner),
  36               m_threadRun(false), m_csdkLock(csdkLock),
  37               m_owner(owner),
  38               m_cfg { cfg }
  39     {
  40         // if the config type is server, we ought to never get called.  If the config type
  41         // is both, we count on the server to run the thread and do the initialize
  42         if(m_cfg.mode == ModeType::Client)
  43         {
  44             OCStackResult result = OCInit(m_cfg.ipAddress.c_str(), m_cfg.port, OC_CLIENT);
  45
  46             if(OC_STACK_OK != result)
  47             {
  48                 throw InitializeException("Error Initializing Stack", result);
  49             }
  50
  51             m_threadRun = true;
  52             m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
  53         }
  54     }
  55
  56     InProcClientWrapper::~InProcClientWrapper()
  57     {
  58         if(m_threadRun && m_listeningThread.joinable())
  59         {
  60             m_threadRun = false;
  61             m_listeningThread.join();
  62         }
  63
  64         OCStop();
  65     }
  66
  67     void InProcClientWrapper::listeningFunc()
  68     {
  69         while(m_threadRun)
  70         {
  71             OCStackResult result;
  72             auto cLock = m_csdkLock.lock();
  73             if(cLock)
  74             {
  75                 std::lock_guard<std::recursive_mutex> lock(*cLock);
  76                 result = OCProcess();
  77             }
  78             else
  79             {
  80                 result = OC_STACK_ERROR;
  81             }
  82
  83             if(result != OC_STACK_OK)
  84             {
  85                 // TODO: do something with result if failed?
  86             }
  87
  88             // To minimize CPU utilization we may wish to do this with sleep
  89             std::this_thread::sleep_for(std::chrono::milliseconds(10));
  90         }
  91     }
  92
  93
  94
  95     std::string convertOCAddrToString(OCDevAddr& addr)
  96     {
  97         // TODO: we currently assume this is a IPV4 address, need to figure out the actual value
  98
  99         uint8_t a, b, c, d;
 100         uint16_t port;
 101
 102         if(OCDevAddrToIPv4Addr(&addr, &a, &b, &c, &d) ==0 && OCDevAddrToPort(&addr, &port)==0)
 103         {
 104             ostringstream os;
 105             os << "coap://"<<(int)a<<'.'<<(int)b<<'.'<<(int)c<<'.'<<(int)d<<':'<<(int)port;
 106             return os.str();
 107         }
 108         else
 109         {
 110             return "INVALID IP";
 111         }
 112     }
 113
 114     struct ListenContext
 115     {
 116         FindCallback              callback;
 117         IClientWrapper::Ptr       clientWrapper;
 118         OC::OCPlatform const*     owner;          // observing ptr
 119     };
 120
 121
 122     const std::string URIKEY = "href";
 123     const std::string OBSERVABLEKEY = "obs";
 124     const std::string RESOURCETYPESKEY= "rt";
 125     const std::string INTERFACESKEY = "if";
 126     const std::string PROPERTYKEY = "prop";
 127     const std::string REPKEY = "rep";
 128
 129     std::shared_ptr<OCResource> InProcClientWrapper::parseOCResource(
 130         IClientWrapper::Ptr clientWrapper, const std::string& host,
 131         const boost::property_tree::ptree resourceNode)
 132     {
 133         std::string uri = resourceNode.get<std::string>(URIKEY, "");
 134         bool obs = resourceNode.get<int>(OBSERVABLEKEY,0) == 1;
 135         std::vector<std::string> rTs;
 136         std::vector<std::string> ifaces;
 137
 138         boost::property_tree::ptree properties =
 139             resourceNode.get_child(PROPERTYKEY, boost::property_tree::ptree());
 140
 141         boost::property_tree::ptree rT =
 142             properties.get_child(RESOURCETYPESKEY, boost::property_tree::ptree());
 143         for(auto itr : rT)
 144         {
 145             rTs.push_back(itr.second.data());
 146         }
 147
 148         boost::property_tree::ptree iF =
 149             properties.get_child(INTERFACESKEY, boost::property_tree::ptree());
 150         for(auto itr : iF)
 151         {
 152             ifaces.push_back(itr.second.data());
 153         }
 154
 155         return std::shared_ptr<OCResource>(
 156             new OCResource(clientWrapper, host, uri, obs, rTs, ifaces));
 157     }
 158
 159     OCStackApplicationResult listenCallback(void* ctx, OCDoHandle handle,
 160         OCClientResponse* clientResponse)
 161     {
 162         ListenContext* context = static_cast<ListenContext*>(ctx);
 163
 164         if(clientResponse->result != OC_STACK_OK)
 165         {
 166             oclog() << "listenCallback(): failed to create resource. clientResponse: "
 167                     << clientResponse->result
 168                     << std::flush;
 169
 170             return OC_STACK_KEEP_TRANSACTION;
 171         }
 172
 173         std::stringstream requestStream;
 174         requestStream << clientResponse->resJSONPayload;
 175
 176         boost::property_tree::ptree root;
 177
 178         try
 179         {
 180                 boost::property_tree::read_json(requestStream, root);
 181         }
 182         catch(boost::property_tree::json_parser::json_parser_error &e)
 183         {
 184                 oclog() << "listenCallback(): read_json() failed: " << e.what()
 185                         << std::flush;
 186
 187                 return OC_STACK_KEEP_TRANSACTION;
 188         }
 189
 190         boost::property_tree::ptree payload =
 191                 root.get_child("oc", boost::property_tree::ptree());
 192
 193         for(auto payloadItr : payload)
 194         {
 195                 try
 196                 {
 197                     std::string host = convertOCAddrToString(*clientResponse->addr);
 198                     std::shared_ptr<OCResource> resource =
 199                         context->clientWrapper->parseOCResource(context->clientWrapper, host,
 200                         payloadItr.second);
 201
 202                     // Note: the call to detach allows the underlying thread to continue until
 203                     // completion and allows us to destroy the exec object. This is apparently NOT
 204                     // a memory leak, as the thread will apparently take care of itself.
 205                     // Additionally, the only parameter here is a shared ptr, so OCResource will be
 206                     // disposed of properly upon completion of the callback handler.
 207                     std::thread exec(context->callback,resource);
 208                     exec.detach();
 209                 }
 210                 catch(ResourceInitException& e)
 211                 {
 212                     oclog() << "listenCallback(): failed to create resource: " << e.what()
 213                             << std::flush;
 214                 }
 215         }
 216
 217         return OC_STACK_KEEP_TRANSACTION;
 218     }
 219
 220     OCStackResult InProcClientWrapper::ListenForResource(const std::string& serviceUrl,
 221         const std::string& resourceType, FindCallback& callback, QualityOfService QoS)
 222     {
 223         OCStackResult result;
 224
 225         OCCallbackData cbdata = {0};
 226
 227         ListenContext* context = new ListenContext();
 228         context->callback = callback;
 229         context->clientWrapper = shared_from_this();
 230         context->owner = &m_owner;
 231
 232         cbdata.context =  static_cast<void*>(context);
 233         cbdata.cb = listenCallback;
 234         cbdata.cd = [](void* c){delete static_cast<ListenContext*>(c);};
 235
 236         auto cLock = m_csdkLock.lock();
 237         if(cLock)
 238         {
 239             std::lock_guard<std::recursive_mutex> lock(*cLock);
 240             OCDoHandle handle;
 241             result = OCDoResource(&handle, OC_REST_GET,
 242                                   resourceType.c_str(),
 243                                   nullptr, nullptr,
 244                                   static_cast<OCQualityOfService>(QoS),
 245                                   &cbdata,
 246                                   NULL, 0);
 247         }
 248         else
 249         {
 250             result = OC_STACK_ERROR;
 251         }
 252         return result;
 253     }
 254
 255     struct GetContext
 256     {
 257         GetCallback callback;
 258     };
 259
 260     struct SetContext
 261     {
 262         PutCallback callback;
 263     };
 264
 265
 266     OCRepresentation parseGetSetCallback(OCClientResponse* clientResponse)
 267     {
 268         std::stringstream requestStream;
 269         requestStream<<clientResponse->resJSONPayload;
 270         if(strlen((char*)clientResponse->resJSONPayload) == 0)
 271         {
 272             return OCRepresentation();
 273         }
 274
 275         boost::property_tree::ptree root;
 276         try
 277         {
 278             boost::property_tree::read_json(requestStream, root);
 279         }
 280         catch(boost::property_tree::json_parser::json_parser_error &e)
 281         {
 282             return OCRepresentation();
 283         }
 284         boost::property_tree::ptree payload = root.get_child("oc", boost::property_tree::ptree());
 285         OCRepresentation root_resource;
 286         std::vector<OCRepresentation> children;
 287         bool isRoot = true;
 288         for ( auto payloadItr : payload)
 289         {
 290             OCRepresentation child;
 291             try
 292             {
 293                 auto resourceNode = payloadItr.second;
 294                 std::string uri = resourceNode.get<std::string>(URIKEY, "");
 295
 296                 if (isRoot)
 297                 {
 298                     root_resource.setUri(uri);
 299                 }
 300                 else
 301                 {
 302                     child.setUri(uri);
 303                 }
 304
 305                 if( resourceNode.count(PROPERTYKEY) != 0 )
 306                 {
 307                     std::vector<std::string> rTs;
 308                     std::vector<std::string> ifaces;
 309                     boost::property_tree::ptree properties =
 310                         resourceNode.get_child(PROPERTYKEY, boost::property_tree::ptree());
 311
 312                     boost::property_tree::ptree rT =
 313                         properties.get_child(RESOURCETYPESKEY, boost::property_tree::ptree());
 314                     for(auto itr : rT)
 315                     {
 316                         rTs.push_back(itr.second.data());
 317                     }
 318
 319                     boost::property_tree::ptree iF =
 320                         properties.get_child(INTERFACESKEY, boost::property_tree::ptree());
 321                     for(auto itr : iF)
 322                     {
 323                         ifaces.push_back(itr.second.data());
 324                     }
 325                     if (isRoot)
 326                     {
 327                         root_resource.setResourceInterfaces(ifaces);
 328                         root_resource.setResourceTypes(rTs);
 329                     }
 330                     else
 331                     {
 332                         child.setResourceInterfaces(ifaces);
 333                         child.setResourceTypes(rTs);
 334                     }
 335                 }
 336
 337                 if( resourceNode.count(REPKEY) != 0 )
 338                 {
 339                     boost::property_tree::ptree rep =
 340                         resourceNode.get_child(REPKEY, boost::property_tree::ptree());
 341                     AttributeMap attrs;
 342                     for( auto item : rep)
 343                     {
 344                         std::string name = item.first.data();
 345                         std::string value = item.second.data();
 346                         attrs[name] = value;
 347                     }
 348                     if (isRoot)
 349                     {
 350                         root_resource.setAttributeMap(attrs);
 351                     }
 352                     else
 353                     {
 354                         child.setAttributeMap(attrs);
 355                     }
 356                 }
 357
 358                 if (!isRoot)
 359                     children.push_back(child);
 360             }
 361             catch (...)
 362             {
 363                 // TODO
 364             }
 365
 366             isRoot = false;
 367          }
 368
 369          root_resource.setChildren(children);
 370
 371         return root_resource;
 372     }
 373
 374     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 375                     HeaderOptions& serverHeaderOptions)
 376     {
 377         if(clientResponse)
 378         {
 379             // Parse header options from server
 380             uint16_t optionID;
 381             std::string optionData;
 382
 383             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 384             {
 385                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 386                 optionData = reinterpret_cast<const char*>
 387                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 388                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 389                 serverHeaderOptions.push_back(headerOption);
 390             }
 391         }
 392         else
 393         {
 394             // clientResponse is invalid
 395             // TODO check proper logging
 396             std::cout << " Invalid response " << std::endl;
 397         }
 398     }
 399
 400     OCStackApplicationResult getResourceCallback(void* ctx, OCDoHandle handle,
 401         OCClientResponse* clientResponse)
 402     {
 403         GetContext* context = static_cast<GetContext*>(ctx);
 404
 405         OCRepresentation rep;
 406         HeaderOptions serverHeaderOptions;
 407         if(clientResponse->result == OC_STACK_OK)
 408         {
 409             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 410             rep = parseGetSetCallback(clientResponse);
 411         }
 412
 413         std::thread exec(context->callback, serverHeaderOptions, rep, clientResponse->result);
 414         exec.detach();
 415         return OC_STACK_DELETE_TRANSACTION;
 416     }
 417
 418     OCStackResult InProcClientWrapper::GetResourceRepresentation(const std::string& host,
 419         const std::string& uri, const QueryParamsMap& queryParams,
 420         const HeaderOptions& headerOptions, GetCallback& callback,
 421         QualityOfService QoS)
 422     {
 423         OCStackResult result;
 424         OCCallbackData cbdata = {0};
 425
 426         GetContext* ctx = new GetContext();
 427         ctx->callback = callback;
 428         cbdata.context = static_cast<void*>(ctx);
 429         cbdata.cb = &getResourceCallback;
 430         cbdata.cd = [](void* c){delete static_cast<GetContext*>(c);};
 431
 432         auto cLock = m_csdkLock.lock();
 433
 434         if(cLock)
 435         {
 436             std::ostringstream os;
 437             os << host << assembleSetResourceUri(uri, queryParams).c_str();
 438
 439             std::lock_guard<std::recursive_mutex> lock(*cLock);
 440             OCDoHandle handle;
 441             OCHeaderOption options[MAX_HEADER_OPTIONS];
 442
 443             assembleHeaderOptions(options, headerOptions);
 444             result = OCDoResource(&handle, OC_REST_GET, os.str().c_str(),
 445                                   nullptr, nullptr,
 446                                   static_cast<OCQualityOfService>(QoS),
 447                                   &cbdata,
 448                                   options, headerOptions.size());
 449         }
 450         else
 451         {
 452             result = OC_STACK_ERROR;
 453         }
 454         return result;
 455     }
 456
 457
 458     OCStackApplicationResult setResourceCallback(void* ctx, OCDoHandle handle,
 459         OCClientResponse* clientResponse)
 460     {
 461         SetContext* context = static_cast<SetContext*>(ctx);
 462         OCRepresentation attrs;
 463         HeaderOptions serverHeaderOptions;
 464
 465         if (OC_STACK_OK               == clientResponse->result ||
 466             OC_STACK_RESOURCE_CREATED == clientResponse->result ||
 467             OC_STACK_RESOURCE_DELETED == clientResponse->result)
 468         {
 469             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 470             attrs = parseGetSetCallback(clientResponse);
 471         }
 472
 473         std::thread exec(context->callback, serverHeaderOptions, attrs, clientResponse->result);
 474         exec.detach();
 475         return OC_STACK_DELETE_TRANSACTION;
 476     }
 477
 478     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 479         const QueryParamsMap& queryParams)
 480     {
 481         if(uri.back() == '/')
 482         {
 483             uri.resize(uri.size()-1);
 484         }
 485
 486         ostringstream paramsList;
 487         if(queryParams.size() > 0)
 488         {
 489             paramsList << '?';
 490         }
 491
 492         for(auto& param : queryParams)
 493         {
 494             paramsList << param.first <<'='<<param.second<<'&';
 495         }
 496
 497         std::string queryString = paramsList.str();
 498         if(queryString.back() == '&')
 499         {
 500             queryString.resize(queryString.size() - 1);
 501         }
 502
 503         std::string ret = uri + queryString;
 504         return ret;
 505     }
 506
 507     std::string InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 508     {
 509         ostringstream payload;
 510         // TODO need to change the format to "{"oc":[]}"
 511         payload << "{\"oc\":";
 512
 513         payload << rep.getJSONRepresentation();
 514
 515         payload << "}";
 516         return payload.str();
 517     }
 518
 519     OCStackResult InProcClientWrapper::PostResourceRepresentation(const std::string& host,
 520         const std::string& uri, const OCRepresentation& rep,
 521         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 522         PostCallback& callback, QualityOfService QoS)
 523     {
 524         OCStackResult result;
 525         OCCallbackData cbdata = {0};
 526
 527         SetContext* ctx = new SetContext();
 528         ctx->callback = callback;
 529         cbdata.cb = &setResourceCallback;
 530         cbdata.cd = [](void* c){delete static_cast<SetContext*>(c);};
 531         cbdata.context = static_cast<void*>(ctx);
 532
 533         // TODO: in the future the cstack should be combining these two strings!
 534         ostringstream os;
 535         os << host << assembleSetResourceUri(uri, queryParams).c_str();
 536         // TODO: end of above
 537
 538         auto cLock = m_csdkLock.lock();
 539
 540         if(cLock)
 541         {
 542             std::lock_guard<std::recursive_mutex> lock(*cLock);
 543             OCHeaderOption options[MAX_HEADER_OPTIONS];
 544             OCDoHandle handle;
 545
 546             assembleHeaderOptions(options, headerOptions);
 547             result = OCDoResource(&handle, OC_REST_POST,
 548                                   os.str().c_str(), nullptr,
 549                                   assembleSetResourcePayload(rep).c_str(),
 550                                   static_cast<OCQualityOfService>(QoS),
 551                                   &cbdata, options, headerOptions.size());
 552         }
 553         else
 554         {
 555             result = OC_STACK_ERROR;
 556         }
 557
 558         return result;
 559     }
 560
 561
 562     OCStackResult InProcClientWrapper::PutResourceRepresentation(const std::string& host,
 563         const std::string& uri, const OCRepresentation& rep,
 564         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 565         PutCallback& callback, QualityOfService QoS)
 566     {
 567         OCStackResult result;
 568         OCCallbackData cbdata = {0};
 569
 570         SetContext* ctx = new SetContext();
 571         ctx->callback = callback;
 572         cbdata.cb = &setResourceCallback;
 573         cbdata.cd = [](void* c){delete static_cast<SetContext*>(c);};
 574         cbdata.context = static_cast<void*>(ctx);
 575
 576         // TODO: in the future the cstack should be combining these two strings!
 577         ostringstream os;
 578         os << host << assembleSetResourceUri(uri, queryParams).c_str();
 579         // TODO: end of above
 580
 581         auto cLock = m_csdkLock.lock();
 582
 583         if(cLock)
 584         {
 585             std::lock_guard<std::recursive_mutex> lock(*cLock);
 586             OCDoHandle handle;
 587             OCHeaderOption options[MAX_HEADER_OPTIONS];
 588
 589             assembleHeaderOptions(options, headerOptions);
 590             result = OCDoResource(&handle, OC_REST_PUT,
 591                                   os.str().c_str(), nullptr,
 592                                   assembleSetResourcePayload(rep).c_str(),
 593                                   static_cast<OCQualityOfService>(QoS),
 594                                   &cbdata,
 595                                   options, headerOptions.size());
 596         }
 597         else
 598         {
 599             result = OC_STACK_ERROR;
 600         }
 601
 602         return result;
 603     }
 604
 605     struct DeleteContext
 606     {
 607         DeleteCallback callback;
 608     };
 609
 610     OCStackApplicationResult deleteResourceCallback(void* ctx, OCDoHandle handle,
 611         OCClientResponse* clientResponse)
 612     {
 613         DeleteContext* context = static_cast<DeleteContext*>(ctx);
 614         OCRepresentation attrs;
 615         HeaderOptions serverHeaderOptions;
 616
 617         if(clientResponse->result == OC_STACK_OK)
 618         {
 619             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 620         }
 621         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 622         exec.detach();
 623         return OC_STACK_DELETE_TRANSACTION;
 624     }
 625
 626     OCStackResult InProcClientWrapper::DeleteResource(const std::string& host,
 627         const std::string& uri, const HeaderOptions& headerOptions,
 628          DeleteCallback& callback, QualityOfService QoS)
 629     {
 630         OCStackResult result;
 631         OCCallbackData cbdata = {0};
 632
 633         DeleteContext* ctx = new DeleteContext();
 634         ctx->callback = callback;
 635         cbdata.cb = &deleteResourceCallback;
 636         cbdata.cd = [](void* c){delete static_cast<DeleteContext*>(c);};
 637         cbdata.context = static_cast<void*>(ctx);
 638
 639         ostringstream os;
 640         os << host << uri;
 641
 642         auto cLock = m_csdkLock.lock();
 643
 644         if(cLock)
 645         {
 646             OCHeaderOption options[MAX_HEADER_OPTIONS];
 647             OCDoHandle handle;
 648
 649             assembleHeaderOptions(options, headerOptions);
 650
 651             std::lock_guard<std::recursive_mutex> lock(*cLock);
 652
 653             result = OCDoResource(&handle, OC_REST_DELETE,
 654                                   os.str().c_str(), nullptr,
 655                                   nullptr, static_cast<OCQualityOfService>(m_cfg.QoS),
 656                                   &cbdata, options, headerOptions.size());
 657         }
 658         else
 659         {
 660             result = OC_STACK_ERROR;
 661         }
 662
 663         return result;
 664     }
 665
 666     struct ObserveContext
 667     {
 668         ObserveCallback callback;
 669     };
 670
 671     OCStackApplicationResult observeResourceCallback(void* ctx, OCDoHandle handle,
 672         OCClientResponse* clientResponse)
 673     {
 674         ObserveContext* context = static_cast<ObserveContext*>(ctx);
 675         OCRepresentation attrs;
 676         HeaderOptions serverHeaderOptions;
 677         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 678
 679         if(clientResponse->result == OC_STACK_OK)
 680         {
 681             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 682             attrs = parseGetSetCallback(clientResponse);
 683         }
 684         std::thread exec(context->callback, serverHeaderOptions, attrs,
 685                     clientResponse->result, sequenceNumber);
 686         exec.detach();
 687         return OC_STACK_KEEP_TRANSACTION;
 688     }
 689
 690     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 691         const std::string& host, const std::string& uri, const QueryParamsMap& queryParams,
 692         const HeaderOptions& headerOptions, ObserveCallback& callback, QualityOfService QoS)
 693     {
 694         OCStackResult result;
 695         OCCallbackData cbdata = {0};
 696
 697         ObserveContext* ctx = new ObserveContext();
 698         ctx->callback = callback;
 699         cbdata.context = static_cast<void*>(ctx);
 700         cbdata.cb = &observeResourceCallback;
 701         cbdata.cd = [](void* c){delete static_cast<ObserveContext*>(c);};
 702
 703         OCMethod method;
 704         if (observeType == ObserveType::Observe)
 705         {
 706             method = OC_REST_OBSERVE;
 707         }
 708         else if (observeType == ObserveType::ObserveAll)
 709         {
 710             method = OC_REST_OBSERVE_ALL;
 711         }
 712         else
 713         {
 714             method = OC_REST_OBSERVE_ALL;
 715         }
 716
 717         auto cLock = m_csdkLock.lock();
 718
 719         if(cLock)
 720         {
 721             std::ostringstream os;
 722             os << host << assembleSetResourceUri(uri, queryParams).c_str();
 723
 724             std::lock_guard<std::recursive_mutex> lock(*cLock);
 725             OCHeaderOption options[MAX_HEADER_OPTIONS];
 726
 727             assembleHeaderOptions(options, headerOptions);
 728             result = OCDoResource(handle, method,
 729                                   os.str().c_str(), nullptr,
 730                                   nullptr,
 731                                   static_cast<OCQualityOfService>(QoS),
 732                                   &cbdata,
 733                                   options, headerOptions.size());
 734         }
 735         else
 736         {
 737             return OC_STACK_ERROR;
 738         }
 739
 740         return result;
 741     }
 742
 743     OCStackResult InProcClientWrapper::CancelObserveResource(OCDoHandle handle,
 744         const std::string& host, const std::string& uri, const HeaderOptions& headerOptions,
 745         QualityOfService QoS)
 746     {
 747         OCStackResult result;
 748         auto cLock = m_csdkLock.lock();
 749
 750         if(cLock)
 751         {
 752             std::lock_guard<std::recursive_mutex> lock(*cLock);
 753             OCHeaderOption options[MAX_HEADER_OPTIONS];
 754
 755             assembleHeaderOptions(options, headerOptions);
 756             result = OCCancel(handle, static_cast<OCQualityOfService>(QoS), options, headerOptions.size());
 757         }
 758         else
 759         {
 760             result = OC_STACK_ERROR;
 761         }
 762
 763         return result;
 764     }
 765
 766     struct SubscribePresenceContext
 767     {
 768         SubscribeCallback callback;
 769     };
 770
 771     OCStackApplicationResult subscribePresenceCallback(void* ctx, OCDoHandle handle,
 772         OCClientResponse* clientResponse)
 773     {
 774         SubscribePresenceContext* context = static_cast<SubscribePresenceContext*>(ctx);
 775         std::thread exec(context->callback, clientResponse->result, clientResponse->sequenceNumber);
 776
 777         exec.detach();
 778         return OC_STACK_KEEP_TRANSACTION;
 779     }
 780
 781     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 782         const std::string& host, SubscribeCallback& presenceHandler)
 783     {
 784         OCCallbackData cbdata = {0};
 785
 786         SubscribePresenceContext* ctx = new SubscribePresenceContext();
 787         ctx->callback = presenceHandler;
 788         cbdata.cb = &subscribePresenceCallback;
 789         cbdata.context = static_cast<void*>(ctx);
 790         cbdata.cd = [](void* c){delete static_cast<SubscribePresenceContext*>(c);};
 791         auto cLock = m_csdkLock.lock();
 792
 793         std::ostringstream os;
 794
 795         os << host << "/oc/presence";
 796
 797         if(!cLock)
 798             return OC_STACK_ERROR;
 799
 800         return OCDoResource(handle, OC_REST_PRESENCE, os.str().c_str(), nullptr, nullptr,
 801                             OC_LOW_QOS, &cbdata, NULL, 0);
 802     }
 803
 804     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 805     {
 806         OCStackResult result;
 807         auto cLock = m_csdkLock.lock();
 808
 809         if(cLock)
 810         {
 811             std::lock_guard<std::recursive_mutex> lock(*cLock);
 812             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 813         }
 814         else
 815         {
 816             result = OC_STACK_ERROR;
 817         }
 818
 819         return result;
 820     }
 821
 822     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 823     {
 824         qos = m_cfg.QoS;
 825         return OC_STACK_OK;
 826     }
 827
 828     void InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 829            const HeaderOptions& headerOptions)
 830     {
 831         int i = 0;
 832
 833         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 834         {
 835             options[i].protocolID = OC_COAP_ID;
 836             options[i].optionID = static_cast<uint16_t>(it->getOptionID());
 837             options[i].optionLength = (it->getOptionData()).length() + 1;
 838             memcpy(options[i].optionData, (it->getOptionData()).c_str(),
 839                     (it->getOptionData()).length() + 1);
 840             i++;
 841         }
 842     }
 843 }