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         ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 169                                 reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 170         // loop to ensure valid construction of all resources
 171         for(auto resource : container.Resources())
 172         {
 173             std::thread exec(context->callback, resource);
 174             exec.detach();
 175         }
 176
 177
 178         return OC_STACK_KEEP_TRANSACTION;
 179     }
 180
 181     OCStackResult InProcClientWrapper::ListenForResource(
 182             const std::string& serviceUrl,
 183             const std::string& resourceType,
 184             OCConnectivityType connectivityType,
 185             FindCallback& callback, QualityOfService QoS)
 186     {
 187         if(!callback)
 188         {
 189             return OC_STACK_INVALID_PARAM;
 190         }
 191
 192         OCStackResult result;
 193         ostringstream resourceUri;
 194         resourceUri << serviceUrl << resourceType;
 195
 196         ClientCallbackContext::ListenContext* context =
 197             new ClientCallbackContext::ListenContext(callback, shared_from_this());
 198         OCCallbackData cbdata(
 199                 static_cast<void*>(context),
 200                 listenCallback,
 201                 [](void* c){delete static_cast<ClientCallbackContext::ListenContext*>(c);}
 202             );
 203
 204         auto cLock = m_csdkLock.lock();
 205         if(cLock)
 206         {
 207             std::lock_guard<std::recursive_mutex> lock(*cLock);
 208             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 209                                   resourceUri.str().c_str(),
 210                                   nullptr, nullptr, connectivityType,
 211                                   static_cast<OCQualityOfService>(QoS),
 212                                   &cbdata,
 213                                   nullptr, 0);
 214         }
 215         else
 216         {
 217             delete context;
 218             result = OC_STACK_ERROR;
 219         }
 220         return result;
 221     }
 222
 223     OCStackApplicationResult listenDeviceCallback(void* ctx,
 224                                                   OCDoHandle /*handle*/,
 225             OCClientResponse* clientResponse)
 226     {
 227         ClientCallbackContext::DeviceListenContext* context =
 228             static_cast<ClientCallbackContext::DeviceListenContext*>(ctx);
 229
 230         try
 231         {
 232             OCRepresentation rep = parseGetSetCallback(clientResponse);
 233             std::thread exec(context->callback, rep);
 234             exec.detach();
 235         }
 236         catch(OC::OCException& e)
 237         {
 238             oclog() <<"Exception in listenDeviceCallback, ignoring response: "
 239                 <<e.what() <<std::flush;
 240         }
 241
 242         return OC_STACK_KEEP_TRANSACTION;
 243     }
 244
 245     OCStackResult InProcClientWrapper::ListenForDevice(
 246             const std::string& serviceUrl,
 247             const std::string& deviceURI,
 248             OCConnectivityType connectivityType,
 249             FindDeviceCallback& callback,
 250             QualityOfService QoS)
 251     {
 252         if(!callback)
 253         {
 254             return OC_STACK_INVALID_PARAM;
 255         }
 256         OCStackResult result;
 257         ostringstream deviceUri;
 258         deviceUri << serviceUrl << deviceURI;
 259
 260         ClientCallbackContext::DeviceListenContext* context =
 261             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
 262         OCCallbackData cbdata(
 263                 static_cast<void*>(context),
 264                 listenDeviceCallback,
 265                 [](void* c){delete static_cast<ClientCallbackContext::DeviceListenContext*>(c);}
 266                 );
 267
 268         auto cLock = m_csdkLock.lock();
 269         if(cLock)
 270         {
 271             std::lock_guard<std::recursive_mutex> lock(*cLock);
 272             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 273                                   deviceUri.str().c_str(),
 274                                   nullptr, nullptr, connectivityType,
 275                                   static_cast<OCQualityOfService>(QoS),
 276                                   &cbdata,
 277                                   nullptr, 0);
 278         }
 279         else
 280         {
 281             delete context;
 282             result = OC_STACK_ERROR;
 283         }
 284         return result;
 285     }
 286
 287     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 288                     HeaderOptions& serverHeaderOptions)
 289     {
 290         if(clientResponse)
 291         {
 292             // Parse header options from server
 293             uint16_t optionID;
 294             std::string optionData;
 295
 296             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 297             {
 298                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 299                 optionData = reinterpret_cast<const char*>
 300                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 301                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 302                 serverHeaderOptions.push_back(headerOption);
 303             }
 304         }
 305         else
 306         {
 307             // clientResponse is invalid
 308             // TODO check proper logging
 309             std::cout << " Invalid response " << std::endl;
 310         }
 311     }
 312
 313     OCStackApplicationResult getResourceCallback(void* ctx,
 314                                                  OCDoHandle /*handle*/,
 315         OCClientResponse* clientResponse)
 316     {
 317         ClientCallbackContext::GetContext* context =
 318             static_cast<ClientCallbackContext::GetContext*>(ctx);
 319
 320         OCRepresentation rep;
 321         HeaderOptions serverHeaderOptions;
 322         OCStackResult result = clientResponse->result;
 323         if(result == OC_STACK_OK)
 324         {
 325             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 326             try
 327             {
 328                 rep = parseGetSetCallback(clientResponse);
 329             }
 330             catch(OC::OCException& e)
 331             {
 332                 result = e.code();
 333             }
 334         }
 335
 336         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 337         exec.detach();
 338         return OC_STACK_DELETE_TRANSACTION;
 339     }
 340
 341     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 342         const OCDevAddr& devAddr,
 343         const std::string& resourceUri,
 344         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 345         GetCallback& callback, QualityOfService QoS)
 346     {
 347         if(!callback)
 348         {
 349             return OC_STACK_INVALID_PARAM;
 350         }
 351         OCStackResult result;
 352         ClientCallbackContext::GetContext* ctx =
 353             new ClientCallbackContext::GetContext(callback);
 354         OCCallbackData cbdata(
 355                 static_cast<void*>(ctx),
 356                 getResourceCallback,
 357                 [](void* c){delete static_cast<ClientCallbackContext::GetContext*>(c);}
 358                 );
 359
 360         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 361
 362         auto cLock = m_csdkLock.lock();
 363
 364         if(cLock)
 365         {
 366             std::lock_guard<std::recursive_mutex> lock(*cLock);
 367             OCHeaderOption options[MAX_HEADER_OPTIONS];
 368
 369             result = OCDoResource(
 370                                   nullptr, OC_REST_GET,
 371                                   uri.c_str(),
 372                                   &devAddr, nullptr,
 373                                   CT_DEFAULT,
 374                                   static_cast<OCQualityOfService>(QoS),
 375                                   &cbdata,
 376                                   assembleHeaderOptions(options, headerOptions),
 377                                   headerOptions.size());
 378         }
 379         else
 380         {
 381             delete ctx;
 382             result = OC_STACK_ERROR;
 383         }
 384         return result;
 385     }
 386
 387
 388     OCStackApplicationResult setResourceCallback(void* ctx,
 389                                                  OCDoHandle /*handle*/,
 390         OCClientResponse* clientResponse)
 391     {
 392         ClientCallbackContext::SetContext* context =
 393             static_cast<ClientCallbackContext::SetContext*>(ctx);
 394         OCRepresentation attrs;
 395         HeaderOptions serverHeaderOptions;
 396
 397         OCStackResult result = clientResponse->result;
 398         if (OC_STACK_OK               == result ||
 399             OC_STACK_RESOURCE_CREATED == result ||
 400             OC_STACK_RESOURCE_DELETED == result)
 401         {
 402             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 403             try
 404             {
 405                 attrs = parseGetSetCallback(clientResponse);
 406             }
 407             catch(OC::OCException& e)
 408             {
 409                 result = e.code();
 410             }
 411         }
 412
 413         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
 414         exec.detach();
 415         return OC_STACK_DELETE_TRANSACTION;
 416     }
 417
 418     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 419         const QueryParamsMap& queryParams)
 420     {
 421         if(uri.back() == '/')
 422         {
 423             uri.resize(uri.size()-1);
 424         }
 425
 426         ostringstream paramsList;
 427         if(queryParams.size() > 0)
 428         {
 429             paramsList << '?';
 430         }
 431
 432         for(auto& param : queryParams)
 433         {
 434             paramsList << param.first <<'='<<param.second<<';';
 435         }
 436
 437         std::string queryString = paramsList.str();
 438         if(queryString.back() == ';')
 439         {
 440             queryString.resize(queryString.size() - 1);
 441         }
 442
 443         std::string ret = uri + queryString;
 444         return ret;
 445     }
 446
 447     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 448     {
 449         MessageContainer ocInfo;
 450         ocInfo.addRepresentation(rep);
 451         for(const OCRepresentation& r : rep.getChildren())
 452         {
 453             ocInfo.addRepresentation(r);
 454         }
 455
 456         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 457     }
 458
 459     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 460         const OCDevAddr& devAddr,
 461         const std::string& uri,
 462         const OCRepresentation& rep,
 463         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 464         PostCallback& callback, QualityOfService QoS)
 465     {
 466         if(!callback)
 467         {
 468             return OC_STACK_INVALID_PARAM;
 469         }
 470         OCStackResult result;
 471         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 472         OCCallbackData cbdata(
 473                 static_cast<void*>(ctx),
 474                 setResourceCallback,
 475                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 476                 );
 477
 478         std::string url = assembleSetResourceUri(uri, queryParams);
 479
 480         auto cLock = m_csdkLock.lock();
 481
 482         if(cLock)
 483         {
 484             std::lock_guard<std::recursive_mutex> lock(*cLock);
 485             OCHeaderOption options[MAX_HEADER_OPTIONS];
 486
 487             result = OCDoResource(nullptr, OC_REST_POST,
 488                                   url.c_str(), &devAddr,
 489                                   assembleSetResourcePayload(rep),
 490                                   CT_DEFAULT,
 491                                   static_cast<OCQualityOfService>(QoS),
 492                                   &cbdata,
 493                                   assembleHeaderOptions(options, headerOptions),
 494                                   headerOptions.size());
 495         }
 496         else
 497         {
 498             delete ctx;
 499             result = OC_STACK_ERROR;
 500         }
 501
 502         return result;
 503     }
 504
 505     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 506         const OCDevAddr& devAddr,
 507         const std::string& uri,
 508         const OCRepresentation& rep,
 509         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 510         PutCallback& callback, QualityOfService QoS)
 511     {
 512         if(!callback)
 513         {
 514             return OC_STACK_INVALID_PARAM;
 515         }
 516         OCStackResult result;
 517         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 518         OCCallbackData cbdata(
 519                 static_cast<void*>(ctx),
 520                 setResourceCallback,
 521                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 522                 );
 523
 524         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 525
 526         auto cLock = m_csdkLock.lock();
 527
 528         if(cLock)
 529         {
 530             std::lock_guard<std::recursive_mutex> lock(*cLock);
 531             OCDoHandle handle;
 532             OCHeaderOption options[MAX_HEADER_OPTIONS];
 533
 534             result = OCDoResource(&handle, OC_REST_PUT,
 535                                   url.c_str(), &devAddr,
 536                                   assembleSetResourcePayload(rep),
 537                                   CT_DEFAULT,
 538                                   static_cast<OCQualityOfService>(QoS),
 539                                   &cbdata,
 540                                   assembleHeaderOptions(options, headerOptions),
 541                                   headerOptions.size());
 542         }
 543         else
 544         {
 545             delete ctx;
 546             result = OC_STACK_ERROR;
 547         }
 548
 549         return result;
 550     }
 551
 552     OCStackApplicationResult deleteResourceCallback(void* ctx,
 553                                                     OCDoHandle /*handle*/,
 554         OCClientResponse* clientResponse)
 555     {
 556         ClientCallbackContext::DeleteContext* context =
 557             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 558         HeaderOptions serverHeaderOptions;
 559
 560         if(clientResponse->result == OC_STACK_OK)
 561         {
 562             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 563         }
 564         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 565         exec.detach();
 566         return OC_STACK_DELETE_TRANSACTION;
 567     }
 568
 569     OCStackResult InProcClientWrapper::DeleteResource(
 570         const OCDevAddr& devAddr,
 571         const std::string& uri,
 572         const HeaderOptions& headerOptions,
 573         DeleteCallback& callback,
 574         QualityOfService /*QoS*/)
 575     {
 576         if(!callback)
 577         {
 578             return OC_STACK_INVALID_PARAM;
 579         }
 580         OCStackResult result;
 581         ClientCallbackContext::DeleteContext* ctx =
 582             new ClientCallbackContext::DeleteContext(callback);
 583         OCCallbackData cbdata(
 584                 static_cast<void*>(ctx),
 585                 deleteResourceCallback,
 586                 [](void* c){delete static_cast<ClientCallbackContext::DeleteContext*>(c);}
 587                 );
 588
 589         auto cLock = m_csdkLock.lock();
 590
 591         if(cLock)
 592         {
 593             OCHeaderOption options[MAX_HEADER_OPTIONS];
 594
 595             std::lock_guard<std::recursive_mutex> lock(*cLock);
 596
 597             result = OCDoResource(nullptr, OC_REST_DELETE,
 598                                   uri.c_str(), &devAddr,
 599                                   nullptr,
 600                                   CT_DEFAULT,
 601                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 602                                   &cbdata,
 603                                   assembleHeaderOptions(options, headerOptions),
 604                                   headerOptions.size());
 605         }
 606         else
 607         {
 608             delete ctx;
 609             result = OC_STACK_ERROR;
 610         }
 611
 612         return result;
 613     }
 614
 615     OCStackApplicationResult observeResourceCallback(void* ctx,
 616                                                      OCDoHandle /*handle*/,
 617         OCClientResponse* clientResponse)
 618     {
 619         ClientCallbackContext::ObserveContext* context =
 620             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 621         OCRepresentation attrs;
 622         HeaderOptions serverHeaderOptions;
 623         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 624         OCStackResult result = clientResponse->result;
 625         if(clientResponse->result == OC_STACK_OK)
 626         {
 627             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 628             try
 629             {
 630                 attrs = parseGetSetCallback(clientResponse);
 631             }
 632             catch(OC::OCException& e)
 633             {
 634                 result = e.code();
 635             }
 636         }
 637         std::thread exec(context->callback, serverHeaderOptions, attrs,
 638                     result, sequenceNumber);
 639         exec.detach();
 640         if(sequenceNumber == OC_OBSERVE_DEREGISTER)
 641         {
 642             return OC_STACK_DELETE_TRANSACTION;
 643         }
 644         return OC_STACK_KEEP_TRANSACTION;
 645     }
 646
 647     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 648         const OCDevAddr& devAddr,
 649         const std::string& uri,
 650         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 651         ObserveCallback& callback, QualityOfService QoS)
 652     {
 653         if(!callback)
 654         {
 655             return OC_STACK_INVALID_PARAM;
 656         }
 657         OCStackResult result;
 658
 659         ClientCallbackContext::ObserveContext* ctx =
 660             new ClientCallbackContext::ObserveContext(callback);
 661         OCCallbackData cbdata(
 662                 static_cast<void*>(ctx),
 663                 observeResourceCallback,
 664                 [](void* c){delete static_cast<ClientCallbackContext::ObserveContext*>(c);}
 665                 );
 666
 667         OCMethod method;
 668         if (observeType == ObserveType::Observe)
 669         {
 670             method = OC_REST_OBSERVE;
 671         }
 672         else if (observeType == ObserveType::ObserveAll)
 673         {
 674             method = OC_REST_OBSERVE_ALL;
 675         }
 676         else
 677         {
 678             method = OC_REST_OBSERVE_ALL;
 679         }
 680
 681         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 682
 683         auto cLock = m_csdkLock.lock();
 684
 685         if(cLock)
 686         {
 687             std::lock_guard<std::recursive_mutex> lock(*cLock);
 688             OCHeaderOption options[MAX_HEADER_OPTIONS];
 689
 690             result = OCDoResource(handle, method,
 691                                   url.c_str(), &devAddr,
 692                                   nullptr,
 693                                   CT_DEFAULT,
 694                                   static_cast<OCQualityOfService>(QoS),
 695                                   &cbdata,
 696                                   assembleHeaderOptions(options, headerOptions),
 697                                   headerOptions.size());
 698         }
 699         else
 700         {
 701             delete ctx;
 702             return OC_STACK_ERROR;
 703         }
 704
 705         return result;
 706     }
 707
 708     OCStackResult InProcClientWrapper::CancelObserveResource(
 709             OCDoHandle handle,
 710             const std::string& /*host*/,
 711             const std::string& /*uri*/,
 712             const HeaderOptions& headerOptions,
 713             QualityOfService QoS)
 714     {
 715         OCStackResult result;
 716         auto cLock = m_csdkLock.lock();
 717
 718         if(cLock)
 719         {
 720             std::lock_guard<std::recursive_mutex> lock(*cLock);
 721             OCHeaderOption options[MAX_HEADER_OPTIONS];
 722
 723             result = OCCancel(handle,
 724                     static_cast<OCQualityOfService>(QoS),
 725                     assembleHeaderOptions(options, headerOptions),
 726                     headerOptions.size());
 727         }
 728         else
 729         {
 730             result = OC_STACK_ERROR;
 731         }
 732
 733         return result;
 734     }
 735
 736     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 737                                                        OCDoHandle /*handle*/,
 738             OCClientResponse* clientResponse)
 739     {
 740         ClientCallbackContext::SubscribePresenceContext* context =
 741         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 742
 743         /*
 744          * This a hack while we rethink presence subscription.
 745          */
 746         std::string url = clientResponse->devAddr.addr;
 747
 748         std::thread exec(context->callback, clientResponse->result,
 749                     clientResponse->sequenceNumber, url);
 750
 751         exec.detach();
 752
 753         return OC_STACK_KEEP_TRANSACTION;
 754     }
 755
 756     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 757         const std::string& host, const std::string& resourceType,
 758         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 759     {
 760         if(!presenceHandler)
 761         {
 762             return OC_STACK_INVALID_PARAM;
 763         }
 764
 765         ClientCallbackContext::SubscribePresenceContext* ctx =
 766             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 767         OCCallbackData cbdata(
 768                 static_cast<void*>(ctx),
 769                 subscribePresenceCallback,
 770                 [](void* c)
 771                 {delete static_cast<ClientCallbackContext::SubscribePresenceContext*>(c);}
 772                 );
 773
 774         auto cLock = m_csdkLock.lock();
 775
 776         std::ostringstream os;
 777         os << host << OC_RSRVD_PRESENCE_URI;
 778
 779         if(!resourceType.empty())
 780         {
 781             os << "?rt=" << resourceType;
 782         }
 783
 784         if(!cLock)
 785         {
 786             delete ctx;
 787             return OC_STACK_ERROR;
 788         }
 789
 790         return OCDoResource(handle, OC_REST_PRESENCE,
 791                             os.str().c_str(), nullptr,
 792                             nullptr, connectivityType,
 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     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 821            const HeaderOptions& headerOptions)
 822     {
 823         int i = 0;
 824
 825         if( headerOptions.size() == 0)
 826         {
 827             return nullptr;
 828         }
 829
 830         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 831         {
 832             options[i] = OCHeaderOption(OC_COAP_ID,
 833                     it->getOptionID(),
 834                     it->getOptionData().length() + 1,
 835                     reinterpret_cast<const uint8_t*>(it->getOptionData().c_str()));
 836             i++;
 837         }
 838
 839         return options;
 840     }
 841 }