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_REPRESENTATION
 105                 )
 106           )
 107         {
 108             //OCPayloadDestroy(clientResponse->payload);
 109             return OCRepresentation();
 110         }
 111
 112         MessageContainer oc;
 113         oc.setPayload(clientResponse->payload);
 114         //OCPayloadDestroy(clientResponse->payload);
 115
 116         std::vector<OCRepresentation>::const_iterator it = oc.representations().begin();
 117         if (it == oc.representations().end())
 118         {
 119             return OCRepresentation();
 120         }
 121
 122         // first one is considered the root, everything else is considered a child of this one.
 123         OCRepresentation root = *it;
 124         root.setDevAddr(clientResponse->devAddr);
 125         root.setUri(clientResponse->resourceUri);
 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         try
 167         {
 168             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 169                                     reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 170             // loop to ensure valid construction of all resources
 171
 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         {
 180             oclog() << "Exception in listCallback, ignoring response: "
 181                     << e.what() << std::flush;
 182         }
 183
 184
 185         return OC_STACK_KEEP_TRANSACTION;
 186     }
 187
 188     OCStackApplicationResult listenErrorCallback(void* ctx, OCDoHandle /*handle*/,
 189         OCClientResponse* clientResponse)
 190     {
 191         if (!ctx || !clientResponse)
 192         {
 193             return OC_STACK_KEEP_TRANSACTION;
 194         }
 195
 196         ClientCallbackContext::ListenErrorContext* context =
 197             static_cast<ClientCallbackContext::ListenErrorContext*>(ctx);
 198         if (!context)
 199         {
 200             return OC_STACK_KEEP_TRANSACTION;
 201         }
 202
 203         OCStackResult result = clientResponse->result;
 204         if (result == OC_STACK_OK)
 205         {
 206             if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 207             {
 208                 oclog() << "listenCallback(): clientResponse payload was null or the wrong type"
 209                     << std::flush;
 210                 return OC_STACK_KEEP_TRANSACTION;
 211             }
 212
 213             auto clientWrapper = context->clientWrapper.lock();
 214
 215             if (!clientWrapper)
 216             {
 217                 oclog() << "listenCallback(): failed to get a shared_ptr to the client wrapper"
 218                         << std::flush;
 219                 return OC_STACK_KEEP_TRANSACTION;
 220             }
 221
 222             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 223                                         reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 224             // loop to ensure valid construction of all resources
 225             for (auto resource : container.Resources())
 226             {
 227                 std::thread exec(context->callback, resource);
 228                 exec.detach();
 229             }
 230             return OC_STACK_KEEP_TRANSACTION;
 231         }
 232
 233         std::string resourceURI = clientResponse->resourceUri;
 234         std::thread exec(context->errorCallback, resourceURI, result);
 235         exec.detach();
 236         return OC_STACK_KEEP_TRANSACTION;
 237     }
 238
 239     OCStackResult InProcClientWrapper::ListenForResource(
 240             const std::string& serviceUrl,
 241             const std::string& resourceType,
 242             OCConnectivityType connectivityType,
 243             FindCallback& callback, QualityOfService QoS)
 244     {
 245         if (!callback)
 246         {
 247             return OC_STACK_INVALID_PARAM;
 248         }
 249
 250         OCStackResult result;
 251         ostringstream resourceUri;
 252         resourceUri << serviceUrl << resourceType;
 253
 254         ClientCallbackContext::ListenContext* context =
 255             new ClientCallbackContext::ListenContext(callback, shared_from_this());
 256         OCCallbackData cbdata;
 257         cbdata.context = static_cast<void*>(context),
 258         cbdata.cb      = listenCallback;
 259         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext*)c;};
 260
 261         auto cLock = m_csdkLock.lock();
 262         if (cLock)
 263         {
 264             std::lock_guard<std::recursive_mutex> lock(*cLock);
 265             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 266                                   resourceUri.str().c_str(),
 267                                   nullptr, nullptr, connectivityType,
 268                                   static_cast<OCQualityOfService>(QoS),
 269                                   &cbdata,
 270                                   nullptr, 0);
 271         }
 272         else
 273         {
 274             delete context;
 275             result = OC_STACK_ERROR;
 276         }
 277         return result;
 278     }
 279
 280     OCStackResult InProcClientWrapper::ListenErrorForResource(
 281             const std::string& serviceUrl,
 282             const std::string& resourceType,
 283             OCConnectivityType connectivityType,
 284             FindCallback& callback, FindErrorCallback& errorCallback,
 285             QualityOfService QoS)
 286     {
 287         if (!callback)
 288         {
 289             return OC_STACK_INVALID_PARAM;
 290         }
 291
 292         ostringstream resourceUri;
 293         resourceUri << serviceUrl << resourceType;
 294
 295         ClientCallbackContext::ListenErrorContext* context =
 296             new ClientCallbackContext::ListenErrorContext(callback, errorCallback,
 297                                                           shared_from_this());
 298         if (!context)
 299         {
 300             return OC_STACK_ERROR;
 301         }
 302
 303         OCCallbackData cbdata(
 304                 static_cast<void*>(context),
 305                 listenErrorCallback,
 306                 [](void* c){delete static_cast<ClientCallbackContext::ListenErrorContext*>(c);}
 307             );
 308
 309         OCStackResult result;
 310         auto cLock = m_csdkLock.lock();
 311         if (cLock)
 312         {
 313             std::lock_guard<std::recursive_mutex> lock(*cLock);
 314             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 315                                   resourceUri.str().c_str(),
 316                                   nullptr, nullptr, connectivityType,
 317                                   static_cast<OCQualityOfService>(QoS),
 318                                   &cbdata,
 319                                   nullptr, 0);
 320         }
 321         else
 322         {
 323             delete context;
 324             result = OC_STACK_ERROR;
 325         }
 326         return result;
 327     }
 328
 329     OCStackApplicationResult listenCallback2(void* ctx, OCDoHandle /*handle*/,
 330         OCClientResponse* clientResponse)
 331     {
 332         ClientCallbackContext::ListenContext2* context =
 333             static_cast<ClientCallbackContext::ListenContext2*>(ctx);
 334
 335         if (clientResponse->result != OC_STACK_OK)
 336         {
 337             oclog() << "listenCallback2(): failed to create resource. clientResponse: "
 338                     << clientResponse->result
 339                     << std::flush;
 340
 341             return OC_STACK_KEEP_TRANSACTION;
 342         }
 343
 344         if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 345         {
 346             oclog() << "listenCallback2(): clientResponse payload was null or the wrong type"
 347                 << std::flush;
 348             return OC_STACK_KEEP_TRANSACTION;
 349         }
 350
 351         auto clientWrapper = context->clientWrapper.lock();
 352
 353         if (!clientWrapper)
 354         {
 355             oclog() << "listenCallback2(): failed to get a shared_ptr to the client wrapper"
 356                     << std::flush;
 357             return OC_STACK_KEEP_TRANSACTION;
 358         }
 359
 360         try
 361         {
 362             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 363                                     reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 364
 365             std::thread exec(context->callback, container.Resources());
 366             exec.detach();
 367         }
 368         catch (std::exception &e)
 369         {
 370             oclog() << "Exception in listCallback2, ignoring response: "
 371                     << e.what() << std::flush;
 372         }
 373
 374
 375         return OC_STACK_KEEP_TRANSACTION;
 376     }
 377
 378     OCStackResult InProcClientWrapper::ListenForResource2(
 379             const std::string& serviceUrl,
 380             const std::string& resourceType,
 381             OCConnectivityType connectivityType,
 382             FindResListCallback& callback, QualityOfService QoS)
 383     {
 384         if (!callback)
 385         {
 386             return OC_STACK_INVALID_PARAM;
 387         }
 388
 389         OCStackResult result;
 390         ostringstream resourceUri;
 391         resourceUri << serviceUrl << resourceType;
 392
 393         ClientCallbackContext::ListenContext2* context =
 394             new ClientCallbackContext::ListenContext2(callback, shared_from_this());
 395         OCCallbackData cbdata;
 396         cbdata.context = static_cast<void*>(context),
 397         cbdata.cb      = listenCallback2;
 398         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext2*)c;};
 399
 400         auto cLock = m_csdkLock.lock();
 401         if (cLock)
 402         {
 403             std::lock_guard<std::recursive_mutex> lock(*cLock);
 404             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 405                                   resourceUri.str().c_str(),
 406                                   nullptr, nullptr, connectivityType,
 407                                   static_cast<OCQualityOfService>(QoS),
 408                                   &cbdata,
 409                                   nullptr, 0);
 410         }
 411         else
 412         {
 413             delete context;
 414             result = OC_STACK_ERROR;
 415         }
 416         return result;
 417     }
 418
 419 #ifdef WITH_MQ
 420     OCStackApplicationResult listenMQCallback(void* ctx, OCDoHandle /*handle*/,
 421                                               OCClientResponse* clientResponse)
 422     {
 423         ClientCallbackContext::MQTopicContext* context =
 424             static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
 425
 426         if (!clientResponse || !context)
 427         {
 428             return OC_STACK_DELETE_TRANSACTION;
 429         }
 430
 431         std::string resourceURI = clientResponse->resourceUri;
 432         if (clientResponse->result != OC_STACK_OK)
 433         {
 434             oclog() << "listenMQCallback(): failed to create resource. clientResponse: "
 435                     << clientResponse->result
 436                     << std::flush;
 437
 438             std::thread exec(context->callback, clientResponse->result,
 439                              resourceURI, nullptr);
 440             exec.detach();
 441
 442             return OC_STACK_DELETE_TRANSACTION;
 443         }
 444
 445         auto clientWrapper = context->clientWrapper.lock();
 446         if (!clientWrapper)
 447         {
 448             oclog() << "listenMQCallback(): failed to get a shared_ptr to the client wrapper"
 449                     << std::flush;
 450             return OC_STACK_DELETE_TRANSACTION;
 451         }
 452
 453         try
 454         {
 455             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 456                                         (OCRepPayload *) clientResponse->payload);
 457
 458             // loop to ensure valid construction of all resources
 459             for (auto resource : container.Resources())
 460             {
 461                 std::thread exec(context->callback, clientResponse->result,
 462                                  resourceURI, resource);
 463                 exec.detach();
 464             }
 465         }
 466         catch (std::exception &e)
 467         {
 468             oclog() << "Exception in listCallback, ignoring response: "
 469                     << e.what() << std::flush;
 470         }
 471
 472         return OC_STACK_DELETE_TRANSACTION;
 473     }
 474
 475     OCStackResult InProcClientWrapper::ListenForMQTopic(const OCDevAddr& devAddr,
 476                                                         const std::string& resourceUri,
 477                                                         const QueryParamsMap& queryParams,
 478                                                         const HeaderOptions& headerOptions,
 479                                                         MQTopicCallback& callback,
 480                                                         QualityOfService QoS)
 481     {
 482         oclog() << "ListenForMQTopic()" << std::flush;
 483
 484         if (!callback)
 485         {
 486             return OC_STACK_INVALID_PARAM;
 487         }
 488
 489         ClientCallbackContext::MQTopicContext* context =
 490             new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 491         OCCallbackData cbdata;
 492         cbdata.context = static_cast<void*>(context),
 493         cbdata.cb      = listenMQCallback;
 494         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 495
 496         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 497
 498         OCStackResult result = OC_STACK_ERROR;
 499         auto cLock = m_csdkLock.lock();
 500         if (cLock)
 501         {
 502             std::lock_guard<std::recursive_mutex> lock(*cLock);
 503             OCHeaderOption options[MAX_HEADER_OPTIONS];
 504             result = OCDoResource(
 505                                   nullptr, OC_REST_GET,
 506                                   uri.c_str(),
 507                                   &devAddr, nullptr,
 508                                   CT_DEFAULT,
 509                                   static_cast<OCQualityOfService>(QoS),
 510                                   &cbdata,
 511                                   assembleHeaderOptions(options, headerOptions),
 512                                   headerOptions.size());
 513         }
 514         else
 515         {
 516             delete context;
 517         }
 518
 519         return result;
 520     }
 521 #endif
 522
 523     OCStackApplicationResult listenDeviceCallback(void* ctx,
 524                                                   OCDoHandle /*handle*/,
 525             OCClientResponse* clientResponse)
 526     {
 527         ClientCallbackContext::DeviceListenContext* context =
 528             static_cast<ClientCallbackContext::DeviceListenContext*>(ctx);
 529
 530         try
 531         {
 532             OCRepresentation rep = parseGetSetCallback(clientResponse);
 533             std::thread exec(context->callback, rep);
 534             exec.detach();
 535         }
 536         catch(OC::OCException& e)
 537         {
 538             oclog() <<"Exception in listenDeviceCallback, ignoring response: "
 539                 <<e.what() <<std::flush;
 540         }
 541
 542         return OC_STACK_KEEP_TRANSACTION;
 543     }
 544
 545     OCStackResult InProcClientWrapper::ListenForDevice(
 546             const std::string& serviceUrl,
 547             const std::string& deviceURI,
 548             OCConnectivityType connectivityType,
 549             FindDeviceCallback& callback,
 550             QualityOfService QoS)
 551     {
 552         if (!callback)
 553         {
 554             return OC_STACK_INVALID_PARAM;
 555         }
 556         OCStackResult result;
 557         ostringstream deviceUri;
 558         deviceUri << serviceUrl << deviceURI;
 559
 560         ClientCallbackContext::DeviceListenContext* context =
 561             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
 562         OCCallbackData cbdata;
 563
 564         cbdata.context = static_cast<void*>(context),
 565         cbdata.cb      = listenDeviceCallback;
 566         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
 567
 568         auto cLock = m_csdkLock.lock();
 569         if (cLock)
 570         {
 571             std::lock_guard<std::recursive_mutex> lock(*cLock);
 572             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 573                                   deviceUri.str().c_str(),
 574                                   nullptr, nullptr, connectivityType,
 575                                   static_cast<OCQualityOfService>(QoS),
 576                                   &cbdata,
 577                                   nullptr, 0);
 578         }
 579         else
 580         {
 581             delete context;
 582             result = OC_STACK_ERROR;
 583         }
 584         return result;
 585     }
 586
 587     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 588                     HeaderOptions& serverHeaderOptions)
 589     {
 590         if (clientResponse)
 591         {
 592             // Parse header options from server
 593             uint16_t optionID;
 594             std::string optionData;
 595
 596             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 597             {
 598                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 599                 optionData = reinterpret_cast<const char*>
 600                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 601                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 602                 serverHeaderOptions.push_back(headerOption);
 603             }
 604         }
 605         else
 606         {
 607             // clientResponse is invalid
 608             // TODO check proper logging
 609             std::cout << " Invalid response " << std::endl;
 610         }
 611     }
 612
 613 #ifdef WITH_MQ
 614     OCStackApplicationResult createMQTopicCallback(void* ctx, OCDoHandle /*handle*/,
 615                     OCClientResponse* clientResponse)
 616     {
 617         ClientCallbackContext::MQTopicContext* context =
 618             static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
 619         HeaderOptions serverHeaderOptions;
 620
 621         if (!clientResponse || !context)
 622         {
 623             return OC_STACK_DELETE_TRANSACTION;
 624         }
 625
 626         std::string createdUri;
 627         bool isLocationOption = false;
 628         OCStackResult result = clientResponse->result;
 629         if (OC_STACK_OK               == result ||
 630             OC_STACK_RESOURCE_CREATED == result)
 631         {
 632             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 633
 634             for (auto headerOption : serverHeaderOptions)
 635             {
 636                 if (HeaderOption::LOCATION_PATH_OPTION_ID == headerOption.getOptionID())
 637                 {
 638                     createdUri += "/";
 639                     createdUri += headerOption.getOptionData();
 640                     if (!isLocationOption)
 641                     {
 642                         isLocationOption = true;
 643                     }
 644                 }
 645             }
 646         }
 647
 648         if (!isLocationOption)
 649         {
 650             createdUri = std::string(clientResponse->resourceUri);
 651         }
 652
 653         auto clientWrapper = context->clientWrapper.lock();
 654
 655         if (!clientWrapper)
 656         {
 657             oclog() << "createMQTopicCallback(): failed to get a shared_ptr to the client wrapper"
 658                     << std::flush;
 659             return OC_STACK_DELETE_TRANSACTION;
 660         }
 661
 662         try
 663         {
 664             if (OC_STACK_OK               == result ||
 665                 OC_STACK_RESOURCE_CREATED == result)
 666             {
 667                 ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 668                                             createdUri);
 669                 for (auto resource : container.Resources())
 670                 {
 671                     std::thread exec(context->callback, result,
 672                                      createdUri,
 673                                      resource);
 674                     exec.detach();
 675                 }
 676             }
 677             else
 678             {
 679                 std::thread exec(context->callback, result,
 680                                  createdUri,
 681                                  nullptr);
 682                 exec.detach();
 683             }
 684         }
 685         catch (std::exception &e)
 686         {
 687             oclog() << "Exception in createMQTopicCallback, ignoring response: "
 688                     << e.what() << std::flush;
 689         }
 690         return OC_STACK_DELETE_TRANSACTION;
 691     }
 692
 693     OCStackResult InProcClientWrapper::PutMQTopicRepresentation(
 694                 const OCDevAddr& devAddr,
 695                 const std::string& uri,
 696                 const OCRepresentation& rep,
 697                 const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 698                 MQTopicCallback& callback, QualityOfService QoS)
 699     {
 700         if (!callback)
 701         {
 702             return OC_STACK_INVALID_PARAM;
 703         }
 704         OCStackResult result;
 705         ClientCallbackContext::MQTopicContext* ctx =
 706                 new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 707         OCCallbackData cbdata;
 708         cbdata.context = static_cast<void*>(ctx),
 709         cbdata.cb      = createMQTopicCallback;
 710         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 711
 712         std::string url = assembleSetResourceUri(uri, queryParams);
 713
 714         auto cLock = m_csdkLock.lock();
 715
 716         if (cLock)
 717         {
 718             std::lock_guard<std::recursive_mutex> lock(*cLock);
 719             OCHeaderOption options[MAX_HEADER_OPTIONS];
 720
 721             result = OCDoResource(nullptr, OC_REST_PUT,
 722                                   url.c_str(), &devAddr,
 723                                   assembleSetResourcePayload(rep),
 724                                   CT_DEFAULT,
 725                                   static_cast<OCQualityOfService>(QoS),
 726                                   &cbdata,
 727                                   assembleHeaderOptions(options, headerOptions),
 728                                   headerOptions.size());
 729         }
 730         else
 731         {
 732             delete ctx;
 733             result = OC_STACK_ERROR;
 734         }
 735
 736         return result;
 737     }
 738 #endif
 739     OCStackApplicationResult getResourceCallback(void* ctx,
 740                                                  OCDoHandle /*handle*/,
 741         OCClientResponse* clientResponse)
 742     {
 743         ClientCallbackContext::GetContext* context =
 744             static_cast<ClientCallbackContext::GetContext*>(ctx);
 745
 746         OCRepresentation rep;
 747         HeaderOptions serverHeaderOptions;
 748         OCStackResult result = clientResponse->result;
 749
 750         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 751         try
 752         {
 753             rep = parseGetSetCallback(clientResponse);
 754         }
 755         catch(OC::OCException& e)
 756         {
 757             result = e.code();
 758         }
 759
 760         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 761         exec.detach();
 762         return OC_STACK_DELETE_TRANSACTION;
 763     }
 764
 765     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 766         const OCDevAddr& devAddr,
 767         const std::string& resourceUri,
 768         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 769         OCConnectivityType connectivityType,
 770         GetCallback& callback, QualityOfService QoS)
 771     {
 772         if (!callback)
 773         {
 774             return OC_STACK_INVALID_PARAM;
 775         }
 776         OCStackResult result;
 777         ClientCallbackContext::GetContext* ctx =
 778             new ClientCallbackContext::GetContext(callback);
 779         OCCallbackData cbdata;
 780         cbdata.context = static_cast<void*>(ctx),
 781         cbdata.cb      = getResourceCallback;
 782         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 783
 784
 785         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 786
 787         auto cLock = m_csdkLock.lock();
 788
 789         if (cLock)
 790         {
 791             std::lock_guard<std::recursive_mutex> lock(*cLock);
 792             OCHeaderOption options[MAX_HEADER_OPTIONS];
 793
 794             result = OCDoResource(
 795                                   nullptr, OC_REST_GET,
 796                                   uri.c_str(),
 797                                   &devAddr, nullptr,
 798                                   connectivityType,
 799                                   static_cast<OCQualityOfService>(QoS),
 800                                   &cbdata,
 801                                   assembleHeaderOptions(options, headerOptions),
 802                                   headerOptions.size());
 803         }
 804         else
 805         {
 806             delete ctx;
 807             result = OC_STACK_ERROR;
 808         }
 809         return result;
 810     }
 811
 812
 813     OCStackApplicationResult setResourceCallback(void* ctx,
 814                                                  OCDoHandle /*handle*/,
 815         OCClientResponse* clientResponse)
 816     {
 817         ClientCallbackContext::SetContext* context =
 818             static_cast<ClientCallbackContext::SetContext*>(ctx);
 819         OCRepresentation attrs;
 820         HeaderOptions serverHeaderOptions;
 821
 822         OCStackResult result = clientResponse->result;
 823
 824         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 825         try
 826         {
 827             attrs = parseGetSetCallback(clientResponse);
 828         }
 829         catch(OC::OCException& e)
 830         {
 831             result = e.code();
 832         }
 833
 834         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
 835         exec.detach();
 836         return OC_STACK_DELETE_TRANSACTION;
 837     }
 838
 839     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 840         const QueryParamsMap& queryParams)
 841     {
 842         if (!uri.empty())
 843         {
 844             if (uri.back() == '/')
 845             {
 846                 uri.resize(uri.size() - 1);
 847             }
 848         }
 849
 850         ostringstream paramsList;
 851         if (queryParams.size() > 0)
 852         {
 853             paramsList << '?';
 854         }
 855
 856         for (auto& param : queryParams)
 857         {
 858             paramsList << param.first <<'='<<param.second<<';';
 859         }
 860
 861         std::string queryString = paramsList.str();
 862
 863         if (queryString.empty())
 864         {
 865             return uri;
 866         }
 867
 868         if (queryString.back() == ';')
 869         {
 870             queryString.resize(queryString.size() - 1);
 871         }
 872
 873         std::string ret = uri + queryString;
 874         return ret;
 875     }
 876
 877     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 878         const QueryParamsList& queryParams)
 879     {
 880         if (!uri.empty())
 881         {
 882             if (uri.back() == '/')
 883             {
 884                 uri.resize(uri.size() - 1);
 885             }
 886         }
 887
 888         ostringstream paramsList;
 889         if (queryParams.size() > 0)
 890         {
 891             paramsList << '?';
 892         }
 893
 894         for (auto& param : queryParams)
 895         {
 896             for (auto& paramList : param.second)
 897             {
 898                 paramsList << param.first << '=' << paramList << ';';
 899             }
 900         }
 901
 902         std::string queryString = paramsList.str();
 903
 904         if (queryString.empty())
 905         {
 906             return uri;
 907         }
 908
 909         if (queryString.back() == ';')
 910         {
 911             queryString.resize(queryString.size() - 1);
 912         }
 913
 914         std::string ret = uri + queryString;
 915         return ret;
 916     }
 917
 918     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 919     {
 920         MessageContainer ocInfo;
 921         ocInfo.addRepresentation(rep);
 922         for(const OCRepresentation& r : rep.getChildren())
 923         {
 924             ocInfo.addRepresentation(r);
 925         }
 926
 927         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 928     }
 929
 930     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 931         const OCDevAddr& devAddr,
 932         const std::string& uri,
 933         const OCRepresentation& rep,
 934         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 935         OCConnectivityType connectivityType,
 936         PostCallback& callback, QualityOfService QoS)
 937     {
 938         if (!callback)
 939         {
 940             return OC_STACK_INVALID_PARAM;
 941         }
 942         OCStackResult result;
 943         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 944         OCCallbackData cbdata;
 945         cbdata.context = static_cast<void*>(ctx),
 946         cbdata.cb      = setResourceCallback;
 947         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 948
 949
 950         std::string url = assembleSetResourceUri(uri, queryParams);
 951
 952         auto cLock = m_csdkLock.lock();
 953
 954         if (cLock)
 955         {
 956             std::lock_guard<std::recursive_mutex> lock(*cLock);
 957             OCHeaderOption options[MAX_HEADER_OPTIONS];
 958
 959             result = OCDoResource(nullptr, OC_REST_POST,
 960                                   url.c_str(), &devAddr,
 961                                   assembleSetResourcePayload(rep),
 962                                   connectivityType,
 963                                   static_cast<OCQualityOfService>(QoS),
 964                                   &cbdata,
 965                                   assembleHeaderOptions(options, headerOptions),
 966                                   headerOptions.size());
 967         }
 968         else
 969         {
 970             delete ctx;
 971             result = OC_STACK_ERROR;
 972         }
 973
 974         return result;
 975     }
 976
 977     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 978         const OCDevAddr& devAddr,
 979         const std::string& uri,
 980         const OCRepresentation& rep,
 981         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 982         PutCallback& callback, QualityOfService QoS)
 983     {
 984         if (!callback)
 985         {
 986             return OC_STACK_INVALID_PARAM;
 987         }
 988         OCStackResult result;
 989         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 990         OCCallbackData cbdata;
 991         cbdata.context = static_cast<void*>(ctx),
 992         cbdata.cb      = setResourceCallback;
 993         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 994
 995
 996         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 997
 998         auto cLock = m_csdkLock.lock();
 999
1000         if (cLock)
1001         {
1002             std::lock_guard<std::recursive_mutex> lock(*cLock);
1003             OCDoHandle handle;
1004             OCHeaderOption options[MAX_HEADER_OPTIONS];
1005
1006             result = OCDoResource(&handle, OC_REST_PUT,
1007                                   url.c_str(), &devAddr,
1008                                   assembleSetResourcePayload(rep),
1009                                   CT_DEFAULT,
1010                                   static_cast<OCQualityOfService>(QoS),
1011                                   &cbdata,
1012                                   assembleHeaderOptions(options, headerOptions),
1013                                   headerOptions.size());
1014         }
1015         else
1016         {
1017             delete ctx;
1018             result = OC_STACK_ERROR;
1019         }
1020
1021         return result;
1022     }
1023
1024     OCStackApplicationResult deleteResourceCallback(void* ctx,
1025                                                     OCDoHandle /*handle*/,
1026         OCClientResponse* clientResponse)
1027     {
1028         ClientCallbackContext::DeleteContext* context =
1029             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
1030         HeaderOptions serverHeaderOptions;
1031
1032         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1033
1034         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
1035         exec.detach();
1036         return OC_STACK_DELETE_TRANSACTION;
1037     }
1038
1039     OCStackResult InProcClientWrapper::DeleteResource(
1040         const OCDevAddr& devAddr,
1041         const std::string& uri,
1042         const HeaderOptions& headerOptions,
1043         OCConnectivityType connectivityType,
1044         DeleteCallback& callback,
1045         QualityOfService /*QoS*/)
1046     {
1047         if (!callback)
1048         {
1049             return OC_STACK_INVALID_PARAM;
1050         }
1051         OCStackResult result;
1052         ClientCallbackContext::DeleteContext* ctx =
1053             new ClientCallbackContext::DeleteContext(callback);
1054         OCCallbackData cbdata;
1055         cbdata.context = static_cast<void*>(ctx),
1056         cbdata.cb      = deleteResourceCallback;
1057         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
1058
1059
1060         auto cLock = m_csdkLock.lock();
1061
1062         if (cLock)
1063         {
1064             OCHeaderOption options[MAX_HEADER_OPTIONS];
1065
1066             std::lock_guard<std::recursive_mutex> lock(*cLock);
1067
1068             result = OCDoResource(nullptr, OC_REST_DELETE,
1069                                   uri.c_str(), &devAddr,
1070                                   nullptr,
1071                                   connectivityType,
1072                                   static_cast<OCQualityOfService>(m_cfg.QoS),
1073                                   &cbdata,
1074                                   assembleHeaderOptions(options, headerOptions),
1075                                   headerOptions.size());
1076         }
1077         else
1078         {
1079             delete ctx;
1080             result = OC_STACK_ERROR;
1081         }
1082
1083         return result;
1084     }
1085
1086     OCStackApplicationResult observeResourceCallback(void* ctx,
1087                                                      OCDoHandle /*handle*/,
1088         OCClientResponse* clientResponse)
1089     {
1090         ClientCallbackContext::ObserveContext* context =
1091             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
1092         OCRepresentation attrs;
1093         HeaderOptions serverHeaderOptions;
1094         uint32_t sequenceNumber = clientResponse->sequenceNumber;
1095         OCStackResult result = clientResponse->result;
1096
1097         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1098         try
1099         {
1100             attrs = parseGetSetCallback(clientResponse);
1101         }
1102         catch(OC::OCException& e)
1103         {
1104             result = e.code();
1105         }
1106
1107         std::thread exec(context->callback, serverHeaderOptions, attrs,
1108                     result, sequenceNumber);
1109         exec.detach();
1110         if (sequenceNumber == MAX_SEQUENCE_NUMBER + 1)
1111         {
1112             return OC_STACK_DELETE_TRANSACTION;
1113         }
1114
1115         return OC_STACK_KEEP_TRANSACTION;
1116     }
1117
1118     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
1119         const OCDevAddr& devAddr,
1120         const std::string& uri,
1121         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1122         ObserveCallback& callback, QualityOfService QoS)
1123     {
1124         if (!callback)
1125         {
1126             return OC_STACK_INVALID_PARAM;
1127         }
1128         OCStackResult result;
1129
1130         ClientCallbackContext::ObserveContext* ctx =
1131             new ClientCallbackContext::ObserveContext(callback);
1132         OCCallbackData cbdata;
1133         cbdata.context = static_cast<void*>(ctx),
1134         cbdata.cb      = observeResourceCallback;
1135         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1136
1137
1138         OCMethod method;
1139         if (observeType == ObserveType::Observe)
1140         {
1141             method = OC_REST_OBSERVE;
1142         }
1143         else if (observeType == ObserveType::ObserveAll)
1144         {
1145             method = OC_REST_OBSERVE_ALL;
1146         }
1147         else
1148         {
1149             method = OC_REST_OBSERVE_ALL;
1150         }
1151
1152         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1153
1154         auto cLock = m_csdkLock.lock();
1155
1156         if (cLock)
1157         {
1158             std::lock_guard<std::recursive_mutex> lock(*cLock);
1159             OCHeaderOption options[MAX_HEADER_OPTIONS];
1160
1161             result = OCDoResource(handle, method,
1162                                   url.c_str(), &devAddr,
1163                                   nullptr,
1164                                   CT_DEFAULT,
1165                                   static_cast<OCQualityOfService>(QoS),
1166                                   &cbdata,
1167                                   assembleHeaderOptions(options, headerOptions),
1168                                   headerOptions.size());
1169         }
1170         else
1171         {
1172             delete ctx;
1173             return OC_STACK_ERROR;
1174         }
1175
1176         return result;
1177     }
1178
1179     OCStackResult InProcClientWrapper::CancelObserveResource(
1180             OCDoHandle handle,
1181             const std::string& /*host*/,
1182             const std::string& /*uri*/,
1183             const HeaderOptions& headerOptions,
1184             QualityOfService QoS)
1185     {
1186         OCStackResult result;
1187         auto cLock = m_csdkLock.lock();
1188
1189         if (cLock)
1190         {
1191             std::lock_guard<std::recursive_mutex> lock(*cLock);
1192             OCHeaderOption options[MAX_HEADER_OPTIONS];
1193
1194             result = OCCancel(handle,
1195                     static_cast<OCQualityOfService>(QoS),
1196                     assembleHeaderOptions(options, headerOptions),
1197                     headerOptions.size());
1198         }
1199         else
1200         {
1201             result = OC_STACK_ERROR;
1202         }
1203
1204         return result;
1205     }
1206
1207     OCStackApplicationResult subscribePresenceCallback(void* ctx,
1208                                                        OCDoHandle /*handle*/,
1209             OCClientResponse* clientResponse)
1210     {
1211         ClientCallbackContext::SubscribePresenceContext* context =
1212         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
1213
1214         /*
1215          * This a hack while we rethink presence subscription.
1216          */
1217         std::string url = clientResponse->devAddr.addr;
1218
1219         std::thread exec(context->callback, clientResponse->result,
1220                     clientResponse->sequenceNumber, url);
1221
1222         exec.detach();
1223
1224         return OC_STACK_KEEP_TRANSACTION;
1225     }
1226
1227     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
1228         const std::string& host, const std::string& resourceType,
1229         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
1230     {
1231         if (!presenceHandler)
1232         {
1233             return OC_STACK_INVALID_PARAM;
1234         }
1235
1236         ClientCallbackContext::SubscribePresenceContext* ctx =
1237             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
1238         OCCallbackData cbdata;
1239         cbdata.context = static_cast<void*>(ctx),
1240         cbdata.cb      = subscribePresenceCallback;
1241         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
1242
1243
1244         auto cLock = m_csdkLock.lock();
1245
1246         std::ostringstream os;
1247         os << host << OC_RSRVD_PRESENCE_URI;
1248
1249         if (!resourceType.empty())
1250         {
1251             os << "?rt=" << resourceType;
1252         }
1253
1254         if (!cLock)
1255         {
1256             delete ctx;
1257             return OC_STACK_ERROR;
1258         }
1259
1260         return OCDoResource(handle, OC_REST_PRESENCE,
1261                             os.str().c_str(), nullptr,
1262                             nullptr, connectivityType,
1263                             OC_LOW_QOS, &cbdata, NULL, 0);
1264     }
1265
1266     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
1267     {
1268         OCStackResult result;
1269         auto cLock = m_csdkLock.lock();
1270
1271         if (cLock)
1272         {
1273             std::lock_guard<std::recursive_mutex> lock(*cLock);
1274             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
1275         }
1276         else
1277         {
1278             result = OC_STACK_ERROR;
1279         }
1280
1281         return result;
1282     }
1283
1284 #ifdef WITH_CLOUD
1285     OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
1286                                                                const std::string& host,
1287                                                                const std::vector<std::string>& di,
1288                                                                OCConnectivityType connectivityType,
1289                                                                ObserveCallback& callback)
1290     {
1291         if (!callback)
1292         {
1293             return OC_STACK_INVALID_PARAM;
1294         }
1295         OCStackResult result;
1296
1297         ClientCallbackContext::ObserveContext* ctx =
1298             new ClientCallbackContext::ObserveContext(callback);
1299         OCCallbackData cbdata;
1300         cbdata.context = static_cast<void*>(ctx),
1301         cbdata.cb      = observeResourceCallback;
1302         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1303
1304         auto cLock = m_csdkLock.lock();
1305
1306         if (cLock)
1307         {
1308             std::lock_guard<std::recursive_mutex> lock(*cLock);
1309
1310             std::ostringstream os;
1311             os << host << OC_RSRVD_DEVICE_PRESENCE_URI;
1312             QueryParamsList queryParams({{OC_RSRVD_DEVICE_ID, di}});
1313             std::string url = assembleSetResourceUri(os.str(), queryParams);
1314
1315             result = OCDoResource(handle, OC_REST_OBSERVE,
1316                                   url.c_str(), nullptr,
1317                                   nullptr, connectivityType,
1318                                   OC_LOW_QOS, &cbdata,
1319                                   nullptr, 0);
1320         }
1321         else
1322         {
1323             delete ctx;
1324             result = OC_STACK_ERROR;
1325         }
1326
1327         return result;
1328     }
1329 #endif
1330
1331     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
1332     {
1333         qos = m_cfg.QoS;
1334         return OC_STACK_OK;
1335     }
1336
1337     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
1338            const HeaderOptions& headerOptions)
1339     {
1340         int i = 0;
1341
1342         if ( headerOptions.size() == 0)
1343         {
1344             return nullptr;
1345         }
1346
1347         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
1348         {
1349             options[i] = OCHeaderOption();
1350             options[i].protocolID = OC_COAP_ID;
1351             options[i].optionID = it->getOptionID();
1352             options[i].optionLength = it->getOptionData().length() + 1;
1353             strcpy((char*)options[i].optionData, (it->getOptionData().c_str()));
1354             i++;
1355         }
1356
1357         return options;
1358     }
1359
1360     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
1361     {
1362         if (!dev)
1363         {
1364             return nullptr;
1365         }
1366
1367         OCDPDev_t* result = new OCDPDev_t(*dev);
1368         result->prm = new OCPrm_t[dev->prmLen];
1369         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
1370         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
1371     }
1372
1373     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
1374     {
1375         while(list)
1376         {
1377             dpList.push_back(cloneDevice(list));
1378             list = list->next;
1379         }
1380     }
1381
1382     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
1383             GetDirectPairedCallback& callback)
1384     {
1385         if (!callback || 0 == waittime)
1386         {
1387             return OC_STACK_INVALID_PARAM;
1388         }
1389
1390         OCStackResult result = OC_STACK_ERROR;
1391         const OCDPDev_t *list = nullptr;
1392         PairedDevices dpDeviceList;
1393
1394         auto cLock = m_csdkLock.lock();
1395
1396         if (cLock)
1397         {
1398             std::lock_guard<std::recursive_mutex> lock(*cLock);
1399
1400             list = OCDiscoverDirectPairingDevices(waittime);
1401             if (NULL == list)
1402             {
1403                 result = OC_STACK_NO_RESOURCE;
1404                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1405                     << std::flush;
1406             }
1407             else {
1408                 convert(list, dpDeviceList);
1409                 std::thread exec(callback, dpDeviceList);
1410                 exec.detach();
1411                 result = OC_STACK_OK;
1412             }
1413         }
1414         else
1415         {
1416             result = OC_STACK_ERROR;
1417         }
1418
1419         return result;
1420     }
1421
1422     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1423     {
1424         if (!callback)
1425         {
1426             return OC_STACK_INVALID_PARAM;
1427         }
1428
1429         OCStackResult result = OC_STACK_ERROR;
1430         const OCDPDev_t *list = nullptr;
1431         PairedDevices dpDeviceList;
1432
1433         auto cLock = m_csdkLock.lock();
1434
1435         if (cLock)
1436         {
1437             std::lock_guard<std::recursive_mutex> lock(*cLock);
1438
1439             list = OCGetDirectPairedDevices();
1440             if (NULL == list)
1441             {
1442                 result = OC_STACK_NO_RESOURCE;
1443                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1444                     << std::flush;
1445             }
1446             else {
1447                 convert(list, dpDeviceList);
1448                 std::thread exec(callback, dpDeviceList);
1449                 exec.detach();
1450                 result = OC_STACK_OK;
1451             }
1452         }
1453         else
1454         {
1455             result = OC_STACK_ERROR;
1456         }
1457
1458         return result;
1459     }
1460
1461     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1462             OCStackResult result)
1463     {
1464
1465         ClientCallbackContext::DirectPairingContext* context =
1466             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1467
1468         std::thread exec(context->callback, cloneDevice(peer), result);
1469         exec.detach();
1470     }
1471
1472     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1473             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1474     {
1475         if (!peer || !callback)
1476         {
1477             oclog() << "Invalid parameters" << std::flush;
1478             return OC_STACK_INVALID_PARAM;
1479         }
1480
1481         OCStackResult result = OC_STACK_ERROR;
1482         ClientCallbackContext::DirectPairingContext* context =
1483             new ClientCallbackContext::DirectPairingContext(callback);
1484
1485         auto cLock = m_csdkLock.lock();
1486         if (cLock)
1487         {
1488             std::lock_guard<std::recursive_mutex> lock(*cLock);
1489             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1490                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1491         }
1492         else
1493         {
1494             delete context;
1495             result = OC_STACK_ERROR;
1496         }
1497         return result;
1498     }
1499 }