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