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         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 452     }
 453
 454     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 455         const OCDevAddr& devAddr,
 456         const std::string& uri,
 457         const OCRepresentation& rep,
 458         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 459         PostCallback& callback, QualityOfService QoS)
 460     {
 461         if(!callback)
 462         {
 463             return OC_STACK_INVALID_PARAM;
 464         }
 465         OCStackResult result;
 466         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 467         OCCallbackData cbdata(
 468                 static_cast<void*>(ctx),
 469                 setResourceCallback,
 470                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 471                 );
 472
 473         std::string url = assembleSetResourceUri(uri, queryParams);
 474
 475         auto cLock = m_csdkLock.lock();
 476
 477         if(cLock)
 478         {
 479             std::lock_guard<std::recursive_mutex> lock(*cLock);
 480             OCHeaderOption options[MAX_HEADER_OPTIONS];
 481
 482             result = OCDoResource(nullptr, OC_REST_POST,
 483                                   url.c_str(), &devAddr,
 484                                   assembleSetResourcePayload(rep),
 485                                   CT_DEFAULT,
 486                                   static_cast<OCQualityOfService>(QoS),
 487                                   &cbdata,
 488                                   assembleHeaderOptions(options, headerOptions),
 489                                   headerOptions.size());
 490         }
 491         else
 492         {
 493             delete ctx;
 494             result = OC_STACK_ERROR;
 495         }
 496
 497         return result;
 498     }
 499
 500     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 501         const OCDevAddr& devAddr,
 502         const std::string& uri,
 503         const OCRepresentation& rep,
 504         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 505         PutCallback& callback, QualityOfService QoS)
 506     {
 507         if(!callback)
 508         {
 509             return OC_STACK_INVALID_PARAM;
 510         }
 511         OCStackResult result;
 512         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 513         OCCallbackData cbdata(
 514                 static_cast<void*>(ctx),
 515                 setResourceCallback,
 516                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 517                 );
 518
 519         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 520
 521         auto cLock = m_csdkLock.lock();
 522
 523         if(cLock)
 524         {
 525             std::lock_guard<std::recursive_mutex> lock(*cLock);
 526             OCDoHandle handle;
 527             OCHeaderOption options[MAX_HEADER_OPTIONS];
 528
 529             result = OCDoResource(&handle, OC_REST_PUT,
 530                                   url.c_str(), &devAddr,
 531                                   assembleSetResourcePayload(rep),
 532                                   CT_DEFAULT,
 533                                   static_cast<OCQualityOfService>(QoS),
 534                                   &cbdata,
 535                                   assembleHeaderOptions(options, headerOptions),
 536                                   headerOptions.size());
 537         }
 538         else
 539         {
 540             delete ctx;
 541             result = OC_STACK_ERROR;
 542         }
 543
 544         return result;
 545     }
 546
 547     OCStackApplicationResult deleteResourceCallback(void* ctx,
 548                                                     OCDoHandle /*handle*/,
 549         OCClientResponse* clientResponse)
 550     {
 551         ClientCallbackContext::DeleteContext* context =
 552             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 553         HeaderOptions serverHeaderOptions;
 554
 555         if(clientResponse->result == OC_STACK_OK)
 556         {
 557             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 558         }
 559         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 560         exec.detach();
 561         return OC_STACK_DELETE_TRANSACTION;
 562     }
 563
 564     OCStackResult InProcClientWrapper::DeleteResource(
 565         const OCDevAddr& devAddr,
 566         const std::string& uri,
 567         const HeaderOptions& headerOptions,
 568         DeleteCallback& callback,
 569         QualityOfService /*QoS*/)
 570     {
 571         if(!callback)
 572         {
 573             return OC_STACK_INVALID_PARAM;
 574         }
 575         OCStackResult result;
 576         ClientCallbackContext::DeleteContext* ctx =
 577             new ClientCallbackContext::DeleteContext(callback);
 578         OCCallbackData cbdata(
 579                 static_cast<void*>(ctx),
 580                 deleteResourceCallback,
 581                 [](void* c){delete static_cast<ClientCallbackContext::DeleteContext*>(c);}
 582                 );
 583
 584         auto cLock = m_csdkLock.lock();
 585
 586         if(cLock)
 587         {
 588             OCHeaderOption options[MAX_HEADER_OPTIONS];
 589
 590             std::lock_guard<std::recursive_mutex> lock(*cLock);
 591
 592             result = OCDoResource(nullptr, OC_REST_DELETE,
 593                                   uri.c_str(), &devAddr,
 594                                   nullptr,
 595                                   CT_DEFAULT,
 596                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 597                                   &cbdata,
 598                                   assembleHeaderOptions(options, headerOptions),
 599                                   headerOptions.size());
 600         }
 601         else
 602         {
 603             delete ctx;
 604             result = OC_STACK_ERROR;
 605         }
 606
 607         return result;
 608     }
 609
 610     OCStackApplicationResult observeResourceCallback(void* ctx,
 611                                                      OCDoHandle /*handle*/,
 612         OCClientResponse* clientResponse)
 613     {
 614         ClientCallbackContext::ObserveContext* context =
 615             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 616         OCRepresentation attrs;
 617         HeaderOptions serverHeaderOptions;
 618         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 619         OCStackResult result = clientResponse->result;
 620         if(clientResponse->result == OC_STACK_OK)
 621         {
 622             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 623             try
 624             {
 625                 attrs = parseGetSetCallback(clientResponse);
 626             }
 627             catch(OC::OCException& e)
 628             {
 629                 result = e.code();
 630             }
 631         }
 632         std::thread exec(context->callback, serverHeaderOptions, attrs,
 633                     result, sequenceNumber);
 634         exec.detach();
 635         if(sequenceNumber == OC_OBSERVE_DEREGISTER)
 636         {
 637             return OC_STACK_DELETE_TRANSACTION;
 638         }
 639         return OC_STACK_KEEP_TRANSACTION;
 640     }
 641
 642     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 643         const OCDevAddr& devAddr,
 644         const std::string& uri,
 645         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 646         ObserveCallback& callback, QualityOfService QoS)
 647     {
 648         if(!callback)
 649         {
 650             return OC_STACK_INVALID_PARAM;
 651         }
 652         OCStackResult result;
 653
 654         ClientCallbackContext::ObserveContext* ctx =
 655             new ClientCallbackContext::ObserveContext(callback);
 656         OCCallbackData cbdata(
 657                 static_cast<void*>(ctx),
 658                 observeResourceCallback,
 659                 [](void* c){delete static_cast<ClientCallbackContext::ObserveContext*>(c);}
 660                 );
 661
 662         OCMethod method;
 663         if (observeType == ObserveType::Observe)
 664         {
 665             method = OC_REST_OBSERVE;
 666         }
 667         else if (observeType == ObserveType::ObserveAll)
 668         {
 669             method = OC_REST_OBSERVE_ALL;
 670         }
 671         else
 672         {
 673             method = OC_REST_OBSERVE_ALL;
 674         }
 675
 676         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 677
 678         auto cLock = m_csdkLock.lock();
 679
 680         if(cLock)
 681         {
 682             std::lock_guard<std::recursive_mutex> lock(*cLock);
 683             OCHeaderOption options[MAX_HEADER_OPTIONS];
 684
 685             result = OCDoResource(handle, method,
 686                                   url.c_str(), &devAddr,
 687                                   nullptr,
 688                                   CT_DEFAULT,
 689                                   static_cast<OCQualityOfService>(QoS),
 690                                   &cbdata,
 691                                   assembleHeaderOptions(options, headerOptions),
 692                                   headerOptions.size());
 693         }
 694         else
 695         {
 696             delete ctx;
 697             return OC_STACK_ERROR;
 698         }
 699
 700         return result;
 701     }
 702
 703     OCStackResult InProcClientWrapper::CancelObserveResource(
 704             OCDoHandle handle,
 705             const std::string& /*host*/,
 706             const std::string& /*uri*/,
 707             const HeaderOptions& headerOptions,
 708             QualityOfService QoS)
 709     {
 710         OCStackResult result;
 711         auto cLock = m_csdkLock.lock();
 712
 713         if(cLock)
 714         {
 715             std::lock_guard<std::recursive_mutex> lock(*cLock);
 716             OCHeaderOption options[MAX_HEADER_OPTIONS];
 717
 718             result = OCCancel(handle,
 719                     static_cast<OCQualityOfService>(QoS),
 720                     assembleHeaderOptions(options, headerOptions),
 721                     headerOptions.size());
 722         }
 723         else
 724         {
 725             result = OC_STACK_ERROR;
 726         }
 727
 728         return result;
 729     }
 730
 731     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 732                                                        OCDoHandle /*handle*/,
 733             OCClientResponse* clientResponse)
 734     {
 735         ClientCallbackContext::SubscribePresenceContext* context =
 736         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 737
 738         /*
 739          * This a hack while we rethink presence subscription.
 740          */
 741         std::string url = clientResponse->devAddr.addr;
 742
 743         std::thread exec(context->callback, clientResponse->result,
 744                     clientResponse->sequenceNumber, url);
 745
 746         exec.detach();
 747
 748         return OC_STACK_KEEP_TRANSACTION;
 749     }
 750
 751     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 752         const std::string& host, const std::string& resourceType,
 753         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 754     {
 755         if(!presenceHandler)
 756         {
 757             return OC_STACK_INVALID_PARAM;
 758         }
 759
 760         ClientCallbackContext::SubscribePresenceContext* ctx =
 761             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 762         OCCallbackData cbdata(
 763                 static_cast<void*>(ctx),
 764                 subscribePresenceCallback,
 765                 [](void* c)
 766                 {delete static_cast<ClientCallbackContext::SubscribePresenceContext*>(c);}
 767                 );
 768
 769         auto cLock = m_csdkLock.lock();
 770
 771         std::ostringstream os;
 772         os << host << OC_RSRVD_PRESENCE_URI;
 773
 774         if(!resourceType.empty())
 775         {
 776             os << "?rt=" << resourceType;
 777         }
 778
 779         if(!cLock)
 780         {
 781             delete ctx;
 782             return OC_STACK_ERROR;
 783         }
 784
 785         return OCDoResource(handle, OC_REST_PRESENCE,
 786                             os.str().c_str(), nullptr,
 787                             nullptr, connectivityType,
 788                             OC_LOW_QOS, &cbdata, NULL, 0);
 789     }
 790
 791     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 792     {
 793         OCStackResult result;
 794         auto cLock = m_csdkLock.lock();
 795
 796         if(cLock)
 797         {
 798             std::lock_guard<std::recursive_mutex> lock(*cLock);
 799             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 800         }
 801         else
 802         {
 803             result = OC_STACK_ERROR;
 804         }
 805
 806         return result;
 807     }
 808
 809     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 810     {
 811         qos = m_cfg.QoS;
 812         return OC_STACK_OK;
 813     }
 814
 815     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 816            const HeaderOptions& headerOptions)
 817     {
 818         int i = 0;
 819
 820         if( headerOptions.size() == 0)
 821         {
 822             return nullptr;
 823         }
 824
 825         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 826         {
 827             options[i] = OCHeaderOption(OC_COAP_ID,
 828                     it->getOptionID(),
 829                     it->getOptionData().length() + 1,
 830                     reinterpret_cast<const uint8_t*>(it->getOptionData().c_str()));
 831             i++;
 832         }
 833
 834         return options;
 835     }
 836 }