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