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