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