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