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             OC_STACK_RESOURCE_CHANGED == result)
 408         {
 409             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 410             try
 411             {
 412                 attrs = parseGetSetCallback(clientResponse);
 413             }
 414             catch(OC::OCException& e)
 415             {
 416                 result = e.code();
 417             }
 418         }
 419
 420         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
 421         exec.detach();
 422         return OC_STACK_DELETE_TRANSACTION;
 423     }
 424
 425     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 426         const QueryParamsMap& queryParams)
 427     {
 428         if (uri.back() == '/')
 429         {
 430             uri.resize(uri.size()-1);
 431         }
 432
 433         ostringstream paramsList;
 434         if (queryParams.size() > 0)
 435         {
 436             paramsList << '?';
 437         }
 438
 439         for(auto& param : queryParams)
 440         {
 441             paramsList << param.first <<'='<<param.second<<';';
 442         }
 443
 444         std::string queryString = paramsList.str();
 445         if (queryString.back() == ';')
 446         {
 447             queryString.resize(queryString.size() - 1);
 448         }
 449
 450         std::string ret = uri + queryString;
 451         return ret;
 452     }
 453
 454     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 455     {
 456         MessageContainer ocInfo;
 457         ocInfo.addRepresentation(rep);
 458         for(const OCRepresentation& r : rep.getChildren())
 459         {
 460             ocInfo.addRepresentation(r);
 461         }
 462
 463         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 464     }
 465
 466     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 467         const OCDevAddr& devAddr,
 468         const std::string& uri,
 469         const OCRepresentation& rep,
 470         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 471         PostCallback& callback, QualityOfService QoS)
 472     {
 473         if (!callback)
 474         {
 475             return OC_STACK_INVALID_PARAM;
 476         }
 477         OCStackResult result;
 478         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 479         OCCallbackData cbdata(
 480                 static_cast<void*>(ctx),
 481                 setResourceCallback,
 482                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 483                 );
 484
 485         std::string url = assembleSetResourceUri(uri, queryParams);
 486
 487         auto cLock = m_csdkLock.lock();
 488
 489         if (cLock)
 490         {
 491             std::lock_guard<std::recursive_mutex> lock(*cLock);
 492             OCHeaderOption options[MAX_HEADER_OPTIONS];
 493
 494             result = OCDoResource(nullptr, OC_REST_POST,
 495                                   url.c_str(), &devAddr,
 496                                   assembleSetResourcePayload(rep),
 497                                   CT_DEFAULT,
 498                                   static_cast<OCQualityOfService>(QoS),
 499                                   &cbdata,
 500                                   assembleHeaderOptions(options, headerOptions),
 501                                   headerOptions.size());
 502         }
 503         else
 504         {
 505             delete ctx;
 506             result = OC_STACK_ERROR;
 507         }
 508
 509         return result;
 510     }
 511
 512     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 513         const OCDevAddr& devAddr,
 514         const std::string& uri,
 515         const OCRepresentation& rep,
 516         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 517         PutCallback& callback, QualityOfService QoS)
 518     {
 519         if (!callback)
 520         {
 521             return OC_STACK_INVALID_PARAM;
 522         }
 523         OCStackResult result;
 524         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 525         OCCallbackData cbdata(
 526                 static_cast<void*>(ctx),
 527                 setResourceCallback,
 528                 [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);}
 529                 );
 530
 531         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 532
 533         auto cLock = m_csdkLock.lock();
 534
 535         if (cLock)
 536         {
 537             std::lock_guard<std::recursive_mutex> lock(*cLock);
 538             OCDoHandle handle;
 539             OCHeaderOption options[MAX_HEADER_OPTIONS];
 540
 541             result = OCDoResource(&handle, OC_REST_PUT,
 542                                   url.c_str(), &devAddr,
 543                                   assembleSetResourcePayload(rep),
 544                                   CT_DEFAULT,
 545                                   static_cast<OCQualityOfService>(QoS),
 546                                   &cbdata,
 547                                   assembleHeaderOptions(options, headerOptions),
 548                                   headerOptions.size());
 549         }
 550         else
 551         {
 552             delete ctx;
 553             result = OC_STACK_ERROR;
 554         }
 555
 556         return result;
 557     }
 558
 559     OCStackApplicationResult deleteResourceCallback(void* ctx,
 560                                                     OCDoHandle /*handle*/,
 561         OCClientResponse* clientResponse)
 562     {
 563         ClientCallbackContext::DeleteContext* context =
 564             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 565         HeaderOptions serverHeaderOptions;
 566
 567         if (clientResponse->result == OC_STACK_OK)
 568         {
 569             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 570         }
 571         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 572         exec.detach();
 573         return OC_STACK_DELETE_TRANSACTION;
 574     }
 575
 576     OCStackResult InProcClientWrapper::DeleteResource(
 577         const OCDevAddr& devAddr,
 578         const std::string& uri,
 579         const HeaderOptions& headerOptions,
 580         DeleteCallback& callback,
 581         QualityOfService /*QoS*/)
 582     {
 583         if (!callback)
 584         {
 585             return OC_STACK_INVALID_PARAM;
 586         }
 587         OCStackResult result;
 588         ClientCallbackContext::DeleteContext* ctx =
 589             new ClientCallbackContext::DeleteContext(callback);
 590         OCCallbackData cbdata(
 591                 static_cast<void*>(ctx),
 592                 deleteResourceCallback,
 593                 [](void* c){delete static_cast<ClientCallbackContext::DeleteContext*>(c);}
 594                 );
 595
 596         auto cLock = m_csdkLock.lock();
 597
 598         if (cLock)
 599         {
 600             OCHeaderOption options[MAX_HEADER_OPTIONS];
 601
 602             std::lock_guard<std::recursive_mutex> lock(*cLock);
 603
 604             result = OCDoResource(nullptr, OC_REST_DELETE,
 605                                   uri.c_str(), &devAddr,
 606                                   nullptr,
 607                                   CT_DEFAULT,
 608                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 609                                   &cbdata,
 610                                   assembleHeaderOptions(options, headerOptions),
 611                                   headerOptions.size());
 612         }
 613         else
 614         {
 615             delete ctx;
 616             result = OC_STACK_ERROR;
 617         }
 618
 619         return result;
 620     }
 621
 622     OCStackApplicationResult observeResourceCallback(void* ctx,
 623                                                      OCDoHandle /*handle*/,
 624         OCClientResponse* clientResponse)
 625     {
 626         ClientCallbackContext::ObserveContext* context =
 627             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 628         OCRepresentation attrs;
 629         HeaderOptions serverHeaderOptions;
 630         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 631         OCStackResult result = clientResponse->result;
 632         if (clientResponse->result == OC_STACK_OK)
 633         {
 634             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 635             try
 636             {
 637                 attrs = parseGetSetCallback(clientResponse);
 638             }
 639             catch(OC::OCException& e)
 640             {
 641                 result = e.code();
 642             }
 643         }
 644         std::thread exec(context->callback, serverHeaderOptions, attrs,
 645                     result, sequenceNumber);
 646         exec.detach();
 647
 648         return OC_STACK_KEEP_TRANSACTION;
 649     }
 650
 651     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 652         const OCDevAddr& devAddr,
 653         const std::string& uri,
 654         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 655         ObserveCallback& callback, QualityOfService QoS)
 656     {
 657         if (!callback)
 658         {
 659             return OC_STACK_INVALID_PARAM;
 660         }
 661         OCStackResult result;
 662
 663         ClientCallbackContext::ObserveContext* ctx =
 664             new ClientCallbackContext::ObserveContext(callback);
 665         OCCallbackData cbdata(
 666                 static_cast<void*>(ctx),
 667                 observeResourceCallback,
 668                 [](void* c){delete static_cast<ClientCallbackContext::ObserveContext*>(c);}
 669                 );
 670
 671         OCMethod method;
 672         if (observeType == ObserveType::Observe)
 673         {
 674             method = OC_REST_OBSERVE;
 675         }
 676         else if (observeType == ObserveType::ObserveAll)
 677         {
 678             method = OC_REST_OBSERVE_ALL;
 679         }
 680         else
 681         {
 682             method = OC_REST_OBSERVE_ALL;
 683         }
 684
 685         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 686
 687         auto cLock = m_csdkLock.lock();
 688
 689         if (cLock)
 690         {
 691             std::lock_guard<std::recursive_mutex> lock(*cLock);
 692             OCHeaderOption options[MAX_HEADER_OPTIONS];
 693
 694             result = OCDoResource(handle, method,
 695                                   url.c_str(), &devAddr,
 696                                   nullptr,
 697                                   CT_DEFAULT,
 698                                   static_cast<OCQualityOfService>(QoS),
 699                                   &cbdata,
 700                                   assembleHeaderOptions(options, headerOptions),
 701                                   headerOptions.size());
 702         }
 703         else
 704         {
 705             delete ctx;
 706             return OC_STACK_ERROR;
 707         }
 708
 709         return result;
 710     }
 711
 712     OCStackResult InProcClientWrapper::CancelObserveResource(
 713             OCDoHandle handle,
 714             const std::string& /*host*/,
 715             const std::string& /*uri*/,
 716             const HeaderOptions& headerOptions,
 717             QualityOfService QoS)
 718     {
 719         OCStackResult result;
 720         auto cLock = m_csdkLock.lock();
 721
 722         if (cLock)
 723         {
 724             std::lock_guard<std::recursive_mutex> lock(*cLock);
 725             OCHeaderOption options[MAX_HEADER_OPTIONS];
 726
 727             result = OCCancel(handle,
 728                     static_cast<OCQualityOfService>(QoS),
 729                     assembleHeaderOptions(options, headerOptions),
 730                     headerOptions.size());
 731         }
 732         else
 733         {
 734             result = OC_STACK_ERROR;
 735         }
 736
 737         return result;
 738     }
 739
 740     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 741                                                        OCDoHandle /*handle*/,
 742             OCClientResponse* clientResponse)
 743     {
 744         ClientCallbackContext::SubscribePresenceContext* context =
 745         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 746
 747         /*
 748          * This a hack while we rethink presence subscription.
 749          */
 750         std::string url = clientResponse->devAddr.addr;
 751
 752         std::thread exec(context->callback, clientResponse->result,
 753                     clientResponse->sequenceNumber, url);
 754
 755         exec.detach();
 756
 757         return OC_STACK_KEEP_TRANSACTION;
 758     }
 759
 760     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 761         const std::string& host, const std::string& resourceType,
 762         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 763     {
 764         if (!presenceHandler)
 765         {
 766             return OC_STACK_INVALID_PARAM;
 767         }
 768
 769         ClientCallbackContext::SubscribePresenceContext* ctx =
 770             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 771         OCCallbackData cbdata(
 772                 static_cast<void*>(ctx),
 773                 subscribePresenceCallback,
 774                 [](void* c)
 775                 {delete static_cast<ClientCallbackContext::SubscribePresenceContext*>(c);}
 776                 );
 777
 778         auto cLock = m_csdkLock.lock();
 779
 780         std::ostringstream os;
 781         os << host << OC_RSRVD_PRESENCE_URI;
 782
 783         if (!resourceType.empty())
 784         {
 785             os << "?rt=" << resourceType;
 786         }
 787
 788         if (!cLock)
 789         {
 790             delete ctx;
 791             return OC_STACK_ERROR;
 792         }
 793
 794         return OCDoResource(handle, OC_REST_PRESENCE,
 795                             os.str().c_str(), nullptr,
 796                             nullptr, connectivityType,
 797                             OC_LOW_QOS, &cbdata, NULL, 0);
 798     }
 799
 800     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 801     {
 802         OCStackResult result;
 803         auto cLock = m_csdkLock.lock();
 804
 805         if (cLock)
 806         {
 807             std::lock_guard<std::recursive_mutex> lock(*cLock);
 808             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 809         }
 810         else
 811         {
 812             result = OC_STACK_ERROR;
 813         }
 814
 815         return result;
 816     }
 817
 818     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 819     {
 820         qos = m_cfg.QoS;
 821         return OC_STACK_OK;
 822     }
 823
 824     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 825            const HeaderOptions& headerOptions)
 826     {
 827         int i = 0;
 828
 829         if ( headerOptions.size() == 0)
 830         {
 831             return nullptr;
 832         }
 833
 834         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 835         {
 836             options[i] = OCHeaderOption(OC_COAP_ID,
 837                     it->getOptionID(),
 838                     it->getOptionData().length() + 1,
 839                     reinterpret_cast<const uint8_t*>(it->getOptionData().c_str()));
 840             i++;
 841         }
 842
 843         return options;
 844     }
 845
 846     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
 847     {
 848         if (!dev)
 849         {
 850             return nullptr;
 851         }
 852
 853         OCDPDev_t* result = new OCDPDev_t(*dev);
 854         result->prm = new OCPrm_t[dev->prmLen];
 855         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
 856         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
 857     }
 858
 859     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
 860     {
 861         while(list)
 862         {
 863             dpList.push_back(cloneDevice(list));
 864             list = list->next;
 865         }
 866     }
 867
 868     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
 869             GetDirectPairedCallback& callback)
 870     {
 871         if (!callback || 0 == waittime)
 872         {
 873             return OC_STACK_INVALID_PARAM;
 874         }
 875
 876         OCStackResult result = OC_STACK_ERROR;
 877         const OCDPDev_t *list = nullptr;
 878         PairedDevices dpDeviceList;
 879
 880         auto cLock = m_csdkLock.lock();
 881
 882         if (cLock)
 883         {
 884             std::lock_guard<std::recursive_mutex> lock(*cLock);
 885
 886             list = OCDiscoverDirectPairingDevices(waittime);
 887             if (NULL == list)
 888             {
 889                 result = OC_STACK_NO_RESOURCE;
 890                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
 891                     << std::flush;
 892             }
 893             else {
 894                 convert(list, dpDeviceList);
 895                 std::thread exec(callback, dpDeviceList);
 896                 exec.detach();
 897                 result = OC_STACK_OK;
 898             }
 899         }
 900         else
 901         {
 902             result = OC_STACK_ERROR;
 903         }
 904
 905         return result;
 906     }
 907
 908     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
 909     {
 910         if (!callback)
 911         {
 912             return OC_STACK_INVALID_PARAM;
 913         }
 914
 915         OCStackResult result = OC_STACK_ERROR;
 916         const OCDPDev_t *list = nullptr;
 917         PairedDevices dpDeviceList;
 918
 919         auto cLock = m_csdkLock.lock();
 920
 921         if (cLock)
 922         {
 923             std::lock_guard<std::recursive_mutex> lock(*cLock);
 924
 925             list = OCGetDirectPairedDevices();
 926             if (NULL == list)
 927             {
 928                 result = OC_STACK_NO_RESOURCE;
 929                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
 930                     << std::flush;
 931             }
 932             else {
 933                 convert(list, dpDeviceList);
 934                 std::thread exec(callback, dpDeviceList);
 935                 exec.detach();
 936                 result = OC_STACK_OK;
 937             }
 938         }
 939         else
 940         {
 941             result = OC_STACK_ERROR;
 942         }
 943
 944         return result;
 945     }
 946
 947     void directPairingCallback(void *ctx, OCDPDev_t *peer,
 948             OCStackResult result)
 949     {
 950
 951         ClientCallbackContext::DirectPairingContext* context =
 952             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
 953
 954         std::thread exec(context->callback, cloneDevice(peer), result);
 955         exec.detach();
 956     }
 957
 958     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
 959             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
 960     {
 961         if (!peer || !callback)
 962         {
 963             oclog() << "Invalid parameters" << std::flush;
 964             return OC_STACK_INVALID_PARAM;
 965         }
 966
 967         OCStackResult result = OC_STACK_ERROR;
 968         ClientCallbackContext::DirectPairingContext* context =
 969             new ClientCallbackContext::DirectPairingContext(callback);
 970
 971         auto cLock = m_csdkLock.lock();
 972         if (cLock)
 973         {
 974             std::lock_guard<std::recursive_mutex> lock(*cLock);
 975             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
 976                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
 977         }
 978         else
 979         {
 980             delete context;
 981             result = OC_STACK_ERROR;
 982         }
 983         return result;
 984     }
 985 }