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