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