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_impl& 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(OC::InitException::STACK_INIT_ERROR, 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 OC::Error::INVALID_IP;
 111         }
 112     }
 113
 114     struct ListenContext
 115     {
 116         FindCallback              callback;
 117         IClientWrapper::Ptr       clientWrapper;
 118     };
 119
 120
 121     std::shared_ptr<OCResource> InProcClientWrapper::parseOCResource(
 122         IClientWrapper::Ptr clientWrapper, const std::string& host,
 123         const boost::property_tree::ptree resourceNode)
 124     {
 125         std::string uri = resourceNode.get<std::string>(OC::Key::URIKEY, "");
 126         bool obs = resourceNode.get<int>(OC::Key::OBSERVABLEKEY,0) == 1;
 127         std::vector<std::string> rTs;
 128         std::vector<std::string> ifaces;
 129
 130         boost::property_tree::ptree properties =
 131             resourceNode.get_child(OC::Key::PROPERTYKEY, boost::property_tree::ptree());
 132
 133         boost::property_tree::ptree rT =
 134             properties.get_child(OC::Key::RESOURCETYPESKEY, boost::property_tree::ptree());
 135         for(auto itr : rT)
 136         {
 137             rTs.push_back(itr.second.data());
 138         }
 139
 140         boost::property_tree::ptree iF =
 141             properties.get_child(OC::Key::INTERFACESKEY, boost::property_tree::ptree());
 142         for(auto itr : iF)
 143         {
 144             ifaces.push_back(itr.second.data());
 145         }
 146
 147         return std::shared_ptr<OCResource>(
 148             new OCResource(clientWrapper, host, uri, obs, rTs, ifaces));
 149     }
 150
 151     OCStackApplicationResult listenCallback(void* ctx, OCDoHandle handle,
 152         OCClientResponse* clientResponse)
 153     {
 154         ListenContext* context = static_cast<ListenContext*>(ctx);
 155
 156         if(clientResponse->result != OC_STACK_OK)
 157         {
 158             oclog() << "listenCallback(): failed to create resource. clientResponse: "
 159                     << clientResponse->result
 160                     << std::flush;
 161
 162             return OC_STACK_KEEP_TRANSACTION;
 163         }
 164
 165         std::stringstream requestStream;
 166         requestStream << clientResponse->resJSONPayload;
 167
 168         boost::property_tree::ptree root;
 169
 170         try
 171         {
 172                 boost::property_tree::read_json(requestStream, root);
 173         }
 174         catch(boost::property_tree::json_parser::json_parser_error &e)
 175         {
 176                 oclog() << "listenCallback(): read_json() failed: " << e.what()
 177                         << std::flush;
 178
 179                 return OC_STACK_KEEP_TRANSACTION;
 180         }
 181
 182         boost::property_tree::ptree payload =
 183                 root.get_child(OC::Key::OCKEY, boost::property_tree::ptree());
 184
 185         for(auto payloadItr : payload)
 186         {
 187                 try
 188                 {
 189                     std::string host = convertOCAddrToString(*clientResponse->addr);
 190                     std::shared_ptr<OCResource> resource =
 191                         context->clientWrapper->parseOCResource(context->clientWrapper, host,
 192                         payloadItr.second);
 193
 194                     // Note: the call to detach allows the underlying thread to continue until
 195                     // completion and allows us to destroy the exec object. This is apparently NOT
 196                     // a memory leak, as the thread will apparently take care of itself.
 197                     // Additionally, the only parameter here is a shared ptr, so OCResource will be
 198                     // disposed of properly upon completion of the callback handler.
 199                     std::thread exec(context->callback,resource);
 200                     exec.detach();
 201                 }
 202                 catch(ResourceInitException& e)
 203                 {
 204                     oclog() << "listenCallback(): failed to create resource: " << e.what()
 205                             << std::flush;
 206                 }
 207         }
 208
 209         return OC_STACK_KEEP_TRANSACTION;
 210     }
 211
 212     OCStackResult InProcClientWrapper::ListenForResource(const std::string& serviceUrl,
 213         const std::string& resourceType, FindCallback& callback, QualityOfService QoS)
 214     {
 215         OCStackResult result;
 216
 217         OCCallbackData cbdata = {0};
 218
 219         ListenContext* context = new ListenContext();
 220         context->callback = callback;
 221         context->clientWrapper = shared_from_this();
 222
 223         cbdata.context =  static_cast<void*>(context);
 224         cbdata.cb = listenCallback;
 225         cbdata.cd = [](void* c){delete static_cast<ListenContext*>(c);};
 226
 227         auto cLock = m_csdkLock.lock();
 228         if(cLock)
 229         {
 230             std::lock_guard<std::recursive_mutex> lock(*cLock);
 231             OCDoHandle handle;
 232             result = OCDoResource(&handle, OC_REST_GET,
 233                                   resourceType.c_str(),
 234                                   nullptr, nullptr,
 235                                   static_cast<OCQualityOfService>(QoS),
 236                                   &cbdata,
 237                                   NULL, 0);
 238         }
 239         else
 240         {
 241             result = OC_STACK_ERROR;
 242         }
 243         return result;
 244     }
 245
 246     struct GetContext
 247     {
 248         GetCallback callback;
 249     };
 250
 251     struct SetContext
 252     {
 253         PutCallback callback;
 254     };
 255
 256
 257     OCRepresentation parseGetSetCallback(OCClientResponse* clientResponse)
 258     {
 259         std::stringstream requestStream;
 260         requestStream<<clientResponse->resJSONPayload;
 261         if(strlen((char*)clientResponse->resJSONPayload) == 0)
 262         {
 263             return OCRepresentation();
 264         }
 265
 266         boost::property_tree::ptree root;
 267         try
 268         {
 269             boost::property_tree::read_json(requestStream, root);
 270         }
 271         catch(boost::property_tree::json_parser::json_parser_error &e)
 272         {
 273             return OCRepresentation();
 274         }
 275         boost::property_tree::ptree payload = root.get_child(OC::Key::OCKEY, boost::property_tree::ptree());
 276         OCRepresentation root_resource;
 277         std::vector<OCRepresentation> children;
 278         bool isRoot = true;
 279         for ( auto payloadItr : payload)
 280         {
 281             OCRepresentation child;
 282             try
 283             {
 284                 auto resourceNode = payloadItr.second;
 285                 std::string uri = resourceNode.get<std::string>(OC::Key::URIKEY, "");
 286
 287                 if (isRoot)
 288                 {
 289                     root_resource.setUri(uri);
 290                 }
 291                 else
 292                 {
 293                     child.setUri(uri);
 294                 }
 295
 296                 if( resourceNode.count(OC::Key::PROPERTYKEY) != 0 )
 297                 {
 298                     std::vector<std::string> rTs;
 299                     std::vector<std::string> ifaces;
 300                     boost::property_tree::ptree properties =
 301                         resourceNode.get_child(OC::Key::PROPERTYKEY, boost::property_tree::ptree());
 302
 303                     boost::property_tree::ptree rT =
 304                         properties.get_child(OC::Key::RESOURCETYPESKEY,
 305                                                 boost::property_tree::ptree());
 306                     for(auto itr : rT)
 307                     {
 308                         rTs.push_back(itr.second.data());
 309                     }
 310
 311                     boost::property_tree::ptree iF =
 312                         properties.get_child(OC::Key::INTERFACESKEY, boost::property_tree::ptree());
 313                     for(auto itr : iF)
 314                     {
 315                         ifaces.push_back(itr.second.data());
 316                     }
 317                     if (isRoot)
 318                     {
 319                         root_resource.setResourceInterfaces(ifaces);
 320                         root_resource.setResourceTypes(rTs);
 321                     }
 322                     else
 323                     {
 324                         child.setResourceInterfaces(ifaces);
 325                         child.setResourceTypes(rTs);
 326                     }
 327                 }
 328
 329                 if( resourceNode.count(OC::Key::REPKEY) != 0 )
 330                 {
 331                     boost::property_tree::ptree rep =
 332                         resourceNode.get_child(OC::Key::REPKEY, boost::property_tree::ptree());
 333                     AttributeMap attrs;
 334                     for( auto item : rep)
 335                     {
 336                         std::string name = item.first.data();
 337                         std::string value = item.second.data();
 338                         attrs[name] = value;
 339                     }
 340                     if (isRoot)
 341                     {
 342                         root_resource.setAttributeMap(attrs);
 343                     }
 344                     else
 345                     {
 346                         child.setAttributeMap(attrs);
 347                     }
 348                 }
 349
 350                 if (!isRoot)
 351                     children.push_back(child);
 352             }
 353             catch (...)
 354             {
 355                 // TODO
 356             }
 357
 358             isRoot = false;
 359          }
 360
 361          root_resource.setChildren(children);
 362
 363         return root_resource;
 364     }
 365
 366     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 367                     HeaderOptions& serverHeaderOptions)
 368     {
 369         if(clientResponse)
 370         {
 371             // Parse header options from server
 372             uint16_t optionID;
 373             std::string optionData;
 374
 375             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 376             {
 377                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 378                 optionData = reinterpret_cast<const char*>
 379                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 380                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 381                 serverHeaderOptions.push_back(headerOption);
 382             }
 383         }
 384         else
 385         {
 386             // clientResponse is invalid
 387             // TODO check proper logging
 388             std::cout << " Invalid response " << std::endl;
 389         }
 390     }
 391
 392     OCStackApplicationResult getResourceCallback(void* ctx, OCDoHandle handle,
 393         OCClientResponse* clientResponse)
 394     {
 395         GetContext* context = static_cast<GetContext*>(ctx);
 396
 397         OCRepresentation rep;
 398         HeaderOptions serverHeaderOptions;
 399         if(clientResponse->result == OC_STACK_OK)
 400         {
 401             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 402             rep = parseGetSetCallback(clientResponse);
 403         }
 404
 405         std::thread exec(context->callback, serverHeaderOptions, rep, clientResponse->result);
 406         exec.detach();
 407         return OC_STACK_DELETE_TRANSACTION;
 408     }
 409
 410     OCStackResult InProcClientWrapper::GetResourceRepresentation(const std::string& host,
 411         const std::string& uri, const QueryParamsMap& queryParams,
 412         const HeaderOptions& headerOptions, GetCallback& callback,
 413         QualityOfService QoS)
 414     {
 415         OCStackResult result;
 416         OCCallbackData cbdata = {0};
 417
 418         GetContext* ctx = new GetContext();
 419         ctx->callback = callback;
 420         cbdata.context = static_cast<void*>(ctx);
 421         cbdata.cb = &getResourceCallback;
 422         cbdata.cd = [](void* c){delete static_cast<GetContext*>(c);};
 423
 424         auto cLock = m_csdkLock.lock();
 425
 426         if(cLock)
 427         {
 428             std::ostringstream os;
 429             os << host << assembleSetResourceUri(uri, queryParams).c_str();
 430
 431             std::lock_guard<std::recursive_mutex> lock(*cLock);
 432             OCDoHandle handle;
 433             OCHeaderOption options[MAX_HEADER_OPTIONS];
 434
 435             assembleHeaderOptions(options, headerOptions);
 436             result = OCDoResource(&handle, OC_REST_GET, os.str().c_str(),
 437                                   nullptr, nullptr,
 438                                   static_cast<OCQualityOfService>(QoS),
 439                                   &cbdata,
 440                                   options, headerOptions.size());
 441         }
 442         else
 443         {
 444             result = OC_STACK_ERROR;
 445         }
 446         return result;
 447     }
 448
 449
 450     OCStackApplicationResult setResourceCallback(void* ctx, OCDoHandle handle,
 451         OCClientResponse* clientResponse)
 452     {
 453         SetContext* context = static_cast<SetContext*>(ctx);
 454         OCRepresentation attrs;
 455         HeaderOptions serverHeaderOptions;
 456
 457         if (OC_STACK_OK               == clientResponse->result ||
 458             OC_STACK_RESOURCE_CREATED == clientResponse->result ||
 459             OC_STACK_RESOURCE_DELETED == clientResponse->result)
 460         {
 461             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 462             attrs = parseGetSetCallback(clientResponse);
 463         }
 464
 465         std::thread exec(context->callback, serverHeaderOptions, attrs, clientResponse->result);
 466         exec.detach();
 467         return OC_STACK_DELETE_TRANSACTION;
 468     }
 469
 470     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 471         const QueryParamsMap& queryParams)
 472     {
 473         if(uri.back() == '/')
 474         {
 475             uri.resize(uri.size()-1);
 476         }
 477
 478         ostringstream paramsList;
 479         if(queryParams.size() > 0)
 480         {
 481             paramsList << '?';
 482         }
 483
 484         for(auto& param : queryParams)
 485         {
 486             paramsList << param.first <<'='<<param.second<<'&';
 487         }
 488
 489         std::string queryString = paramsList.str();
 490         if(queryString.back() == '&')
 491         {
 492             queryString.resize(queryString.size() - 1);
 493         }
 494
 495         std::string ret = uri + queryString;
 496         return ret;
 497     }
 498
 499     std::string InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 500     {
 501         ostringstream payload;
 502         // TODO need to change the format to "{"oc":[]}"
 503         payload << "{\"oc\":";
 504
 505         payload << rep.getJSONRepresentation();
 506
 507         payload << "}";
 508         return payload.str();
 509     }
 510
 511     OCStackResult InProcClientWrapper::PostResourceRepresentation(const std::string& host,
 512         const std::string& uri, const OCRepresentation& rep,
 513         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 514         PostCallback& callback, QualityOfService QoS)
 515     {
 516         OCStackResult result;
 517         OCCallbackData cbdata = {0};
 518
 519         SetContext* ctx = new SetContext();
 520         ctx->callback = callback;
 521         cbdata.cb = &setResourceCallback;
 522         cbdata.cd = [](void* c){delete static_cast<SetContext*>(c);};
 523         cbdata.context = static_cast<void*>(ctx);
 524
 525         // TODO: in the future the cstack should be combining these two strings!
 526         ostringstream os;
 527         os << host << assembleSetResourceUri(uri, queryParams).c_str();
 528         // TODO: end of above
 529
 530         auto cLock = m_csdkLock.lock();
 531
 532         if(cLock)
 533         {
 534             std::lock_guard<std::recursive_mutex> lock(*cLock);
 535             OCHeaderOption options[MAX_HEADER_OPTIONS];
 536             OCDoHandle handle;
 537
 538             assembleHeaderOptions(options, headerOptions);
 539             result = OCDoResource(&handle, OC_REST_POST,
 540                                   os.str().c_str(), nullptr,
 541                                   assembleSetResourcePayload(rep).c_str(),
 542                                   static_cast<OCQualityOfService>(QoS),
 543                                   &cbdata, options, headerOptions.size());
 544         }
 545         else
 546         {
 547             result = OC_STACK_ERROR;
 548         }
 549
 550         return result;
 551     }
 552
 553
 554     OCStackResult InProcClientWrapper::PutResourceRepresentation(const std::string& host,
 555         const std::string& uri, const OCRepresentation& rep,
 556         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 557         PutCallback& callback, QualityOfService QoS)
 558     {
 559         OCStackResult result;
 560         OCCallbackData cbdata = {0};
 561
 562         SetContext* ctx = new SetContext();
 563         ctx->callback = callback;
 564         cbdata.cb = &setResourceCallback;
 565         cbdata.cd = [](void* c){delete static_cast<SetContext*>(c);};
 566         cbdata.context = static_cast<void*>(ctx);
 567
 568         // TODO: in the future the cstack should be combining these two strings!
 569         ostringstream os;
 570         os << host << assembleSetResourceUri(uri, queryParams).c_str();
 571         // TODO: end of above
 572
 573         auto cLock = m_csdkLock.lock();
 574
 575         if(cLock)
 576         {
 577             std::lock_guard<std::recursive_mutex> lock(*cLock);
 578             OCDoHandle handle;
 579             OCHeaderOption options[MAX_HEADER_OPTIONS];
 580
 581             assembleHeaderOptions(options, headerOptions);
 582             result = OCDoResource(&handle, OC_REST_PUT,
 583                                   os.str().c_str(), nullptr,
 584                                   assembleSetResourcePayload(rep).c_str(),
 585                                   static_cast<OCQualityOfService>(QoS),
 586                                   &cbdata,
 587                                   options, headerOptions.size());
 588         }
 589         else
 590         {
 591             result = OC_STACK_ERROR;
 592         }
 593
 594         return result;
 595     }
 596
 597     struct DeleteContext
 598     {
 599         DeleteCallback callback;
 600     };
 601
 602     OCStackApplicationResult deleteResourceCallback(void* ctx, OCDoHandle handle,
 603         OCClientResponse* clientResponse)
 604     {
 605         DeleteContext* context = static_cast<DeleteContext*>(ctx);
 606         OCRepresentation attrs;
 607         HeaderOptions serverHeaderOptions;
 608
 609         if(clientResponse->result == OC_STACK_OK)
 610         {
 611             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 612         }
 613         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 614         exec.detach();
 615         return OC_STACK_DELETE_TRANSACTION;
 616     }
 617
 618     OCStackResult InProcClientWrapper::DeleteResource(const std::string& host,
 619         const std::string& uri, const HeaderOptions& headerOptions,
 620          DeleteCallback& callback, QualityOfService QoS)
 621     {
 622         OCStackResult result;
 623         OCCallbackData cbdata = {0};
 624
 625         DeleteContext* ctx = new DeleteContext();
 626         ctx->callback = callback;
 627         cbdata.cb = &deleteResourceCallback;
 628         cbdata.cd = [](void* c){delete static_cast<DeleteContext*>(c);};
 629         cbdata.context = static_cast<void*>(ctx);
 630
 631         ostringstream os;
 632         os << host << uri;
 633
 634         auto cLock = m_csdkLock.lock();
 635
 636         if(cLock)
 637         {
 638             OCHeaderOption options[MAX_HEADER_OPTIONS];
 639             OCDoHandle handle;
 640
 641             assembleHeaderOptions(options, headerOptions);
 642
 643             std::lock_guard<std::recursive_mutex> lock(*cLock);
 644
 645             result = OCDoResource(&handle, OC_REST_DELETE,
 646                                   os.str().c_str(), nullptr,
 647                                   nullptr, static_cast<OCQualityOfService>(m_cfg.QoS),
 648                                   &cbdata, options, headerOptions.size());
 649         }
 650         else
 651         {
 652             result = OC_STACK_ERROR;
 653         }
 654
 655         return result;
 656     }
 657
 658     struct ObserveContext
 659     {
 660         ObserveCallback callback;
 661     };
 662
 663     OCStackApplicationResult observeResourceCallback(void* ctx, OCDoHandle handle,
 664         OCClientResponse* clientResponse)
 665     {
 666         ObserveContext* context = static_cast<ObserveContext*>(ctx);
 667         OCRepresentation attrs;
 668         HeaderOptions serverHeaderOptions;
 669         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 670
 671         if(clientResponse->result == OC_STACK_OK)
 672         {
 673             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 674             attrs = parseGetSetCallback(clientResponse);
 675         }
 676         std::thread exec(context->callback, serverHeaderOptions, attrs,
 677                     clientResponse->result, sequenceNumber);
 678         exec.detach();
 679         return OC_STACK_KEEP_TRANSACTION;
 680     }
 681
 682     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 683         const std::string& host, const std::string& uri, const QueryParamsMap& queryParams,
 684         const HeaderOptions& headerOptions, ObserveCallback& callback, QualityOfService QoS)
 685     {
 686         OCStackResult result;
 687         OCCallbackData cbdata = {0};
 688
 689         ObserveContext* ctx = new ObserveContext();
 690         ctx->callback = callback;
 691         cbdata.context = static_cast<void*>(ctx);
 692         cbdata.cb = &observeResourceCallback;
 693         cbdata.cd = [](void* c){delete static_cast<ObserveContext*>(c);};
 694
 695         OCMethod method;
 696         if (observeType == ObserveType::Observe)
 697         {
 698             method = OC_REST_OBSERVE;
 699         }
 700         else if (observeType == ObserveType::ObserveAll)
 701         {
 702             method = OC_REST_OBSERVE_ALL;
 703         }
 704         else
 705         {
 706             method = OC_REST_OBSERVE_ALL;
 707         }
 708
 709         auto cLock = m_csdkLock.lock();
 710
 711         if(cLock)
 712         {
 713             std::ostringstream os;
 714             os << host << assembleSetResourceUri(uri, queryParams).c_str();
 715
 716             std::lock_guard<std::recursive_mutex> lock(*cLock);
 717             OCHeaderOption options[MAX_HEADER_OPTIONS];
 718
 719             assembleHeaderOptions(options, headerOptions);
 720             result = OCDoResource(handle, method,
 721                                   os.str().c_str(), nullptr,
 722                                   nullptr,
 723                                   static_cast<OCQualityOfService>(QoS),
 724                                   &cbdata,
 725                                   options, headerOptions.size());
 726         }
 727         else
 728         {
 729             return OC_STACK_ERROR;
 730         }
 731
 732         return result;
 733     }
 734
 735     OCStackResult InProcClientWrapper::CancelObserveResource(OCDoHandle handle,
 736         const std::string& host, const std::string& uri, const HeaderOptions& headerOptions,
 737         QualityOfService QoS)
 738     {
 739         OCStackResult result;
 740         auto cLock = m_csdkLock.lock();
 741
 742         if(cLock)
 743         {
 744             std::lock_guard<std::recursive_mutex> lock(*cLock);
 745             OCHeaderOption options[MAX_HEADER_OPTIONS];
 746
 747             assembleHeaderOptions(options, headerOptions);
 748             result = OCCancel(handle, static_cast<OCQualityOfService>(QoS), options, headerOptions.size());
 749         }
 750         else
 751         {
 752             result = OC_STACK_ERROR;
 753         }
 754
 755         return result;
 756     }
 757
 758     struct SubscribePresenceContext
 759     {
 760         SubscribeCallback callback;
 761     };
 762
 763     OCStackApplicationResult subscribePresenceCallback(void* ctx, OCDoHandle handle,
 764         OCClientResponse* clientResponse)
 765     {
 766         SubscribePresenceContext* context = static_cast<SubscribePresenceContext*>(ctx);
 767         std::thread exec(context->callback, clientResponse->result, clientResponse->sequenceNumber);
 768
 769         exec.detach();
 770         return OC_STACK_KEEP_TRANSACTION;
 771     }
 772
 773     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 774         const std::string& host, SubscribeCallback& presenceHandler)
 775     {
 776         OCCallbackData cbdata = {0};
 777
 778         SubscribePresenceContext* ctx = new SubscribePresenceContext();
 779         ctx->callback = presenceHandler;
 780         cbdata.cb = &subscribePresenceCallback;
 781         cbdata.context = static_cast<void*>(ctx);
 782         cbdata.cd = [](void* c){delete static_cast<SubscribePresenceContext*>(c);};
 783         auto cLock = m_csdkLock.lock();
 784
 785         std::ostringstream os;
 786
 787         os << host << "/oc/presence";
 788
 789         if(!cLock)
 790             return OC_STACK_ERROR;
 791
 792         return OCDoResource(handle, OC_REST_PRESENCE, os.str().c_str(), nullptr, nullptr,
 793                             OC_LOW_QOS, &cbdata, NULL, 0);
 794     }
 795
 796     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 797     {
 798         OCStackResult result;
 799         auto cLock = m_csdkLock.lock();
 800
 801         if(cLock)
 802         {
 803             std::lock_guard<std::recursive_mutex> lock(*cLock);
 804             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 805         }
 806         else
 807         {
 808             result = OC_STACK_ERROR;
 809         }
 810
 811         return result;
 812     }
 813
 814     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 815     {
 816         qos = m_cfg.QoS;
 817         return OC_STACK_OK;
 818     }
 819
 820     void InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 821            const HeaderOptions& headerOptions)
 822     {
 823         int i = 0;
 824
 825         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 826         {
 827             options[i].protocolID = OC_COAP_ID;
 828             options[i].optionID = static_cast<uint16_t>(it->getOptionID());
 829             options[i].optionLength = (it->getOptionData()).length() + 1;
 830             memcpy(options[i].optionData, (it->getOptionData()).c_str(),
 831                     (it->getOptionData()).length() + 1);
 832             i++;
 833         }
 834     }
 835 }