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