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