resource/src/InProcClientWrapper.cpp

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