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     OCStackApplicationResult listenErrorCallback(void* ctx, OCDoHandle /*handle*/,
 188         OCClientResponse* clientResponse)
 189     {
 190         if (!ctx || !clientResponse)
 191         {
 192             return OC_STACK_KEEP_TRANSACTION;
 193         }
 194
 195         ClientCallbackContext::ListenErrorContext* context =
 196             static_cast<ClientCallbackContext::ListenErrorContext*>(ctx);
 197         if (!context)
 198         {
 199             return OC_STACK_KEEP_TRANSACTION;
 200         }
 201
 202         OCStackResult result = clientResponse->result;
 203         if (result == OC_STACK_OK)
 204         {
 205             if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 206             {
 207                 oclog() << "listenCallback(): clientResponse payload was null or the wrong type"
 208                     << std::flush;
 209                 return OC_STACK_KEEP_TRANSACTION;
 210             }
 211
 212             auto clientWrapper = context->clientWrapper.lock();
 213
 214             if (!clientWrapper)
 215             {
 216                 oclog() << "listenCallback(): failed to get a shared_ptr to the client wrapper"
 217                         << std::flush;
 218                 return OC_STACK_KEEP_TRANSACTION;
 219             }
 220
 221             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 222                                         reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 223             // loop to ensure valid construction of all resources
 224             for (auto resource : container.Resources())
 225             {
 226                 std::thread exec(context->callback, resource);
 227                 exec.detach();
 228             }
 229             return OC_STACK_KEEP_TRANSACTION;
 230         }
 231
 232         std::string resourceURI = clientResponse->resourceUri;
 233         std::thread exec(context->errorCallback, resourceURI, result);
 234         exec.detach();
 235         return OC_STACK_DELETE_TRANSACTION;
 236     }
 237
 238     OCStackResult InProcClientWrapper::ListenForResource(
 239             const std::string& serviceUrl,
 240             const std::string& resourceType,
 241             OCConnectivityType connectivityType,
 242             FindCallback& callback, QualityOfService QoS)
 243     {
 244         if (!callback)
 245         {
 246             return OC_STACK_INVALID_PARAM;
 247         }
 248
 249         OCStackResult result;
 250         ostringstream resourceUri;
 251         resourceUri << serviceUrl << resourceType;
 252
 253         ClientCallbackContext::ListenContext* context =
 254             new ClientCallbackContext::ListenContext(callback, shared_from_this());
 255         OCCallbackData cbdata;
 256         cbdata.context = static_cast<void*>(context),
 257         cbdata.cb      = listenCallback;
 258         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext*)c;};
 259
 260         auto cLock = m_csdkLock.lock();
 261         if (cLock)
 262         {
 263             std::lock_guard<std::recursive_mutex> lock(*cLock);
 264             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 265                                   resourceUri.str().c_str(),
 266                                   nullptr, nullptr, connectivityType,
 267                                   static_cast<OCQualityOfService>(QoS),
 268                                   &cbdata,
 269                                   nullptr, 0);
 270         }
 271         else
 272         {
 273             delete context;
 274             result = OC_STACK_ERROR;
 275         }
 276         return result;
 277     }
 278
 279     OCStackResult InProcClientWrapper::ListenErrorForResource(
 280             const std::string& serviceUrl,
 281             const std::string& resourceType,
 282             OCConnectivityType connectivityType,
 283             FindCallback& callback, FindErrorCallback& errorCallback,
 284             QualityOfService QoS)
 285     {
 286         if (!callback)
 287         {
 288             return OC_STACK_INVALID_PARAM;
 289         }
 290
 291         ostringstream resourceUri;
 292         resourceUri << serviceUrl << resourceType;
 293
 294         ClientCallbackContext::ListenErrorContext* context =
 295             new ClientCallbackContext::ListenErrorContext(callback, errorCallback,
 296                                                           shared_from_this());
 297         if (!context)
 298         {
 299             return OC_STACK_ERROR;
 300         }
 301
 302         OCCallbackData cbdata(
 303                 static_cast<void*>(context),
 304                 listenErrorCallback,
 305                 [](void* c){delete static_cast<ClientCallbackContext::ListenErrorContext*>(c);}
 306             );
 307
 308         OCStackResult result;
 309         auto cLock = m_csdkLock.lock();
 310         if (cLock)
 311         {
 312             std::lock_guard<std::recursive_mutex> lock(*cLock);
 313             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 314                                   resourceUri.str().c_str(),
 315                                   nullptr, nullptr, connectivityType,
 316                                   static_cast<OCQualityOfService>(QoS),
 317                                   &cbdata,
 318                                   nullptr, 0);
 319         }
 320         else
 321         {
 322             delete context;
 323             result = OC_STACK_ERROR;
 324         }
 325         return result;
 326     }
 327
 328     OCStackApplicationResult listenDeviceCallback(void* ctx,
 329                                                   OCDoHandle /*handle*/,
 330             OCClientResponse* clientResponse)
 331     {
 332         ClientCallbackContext::DeviceListenContext* context =
 333             static_cast<ClientCallbackContext::DeviceListenContext*>(ctx);
 334
 335         try
 336         {
 337             OCRepresentation rep = parseGetSetCallback(clientResponse);
 338             std::thread exec(context->callback, rep);
 339             exec.detach();
 340         }
 341         catch(OC::OCException& e)
 342         {
 343             oclog() <<"Exception in listenDeviceCallback, ignoring response: "
 344                 <<e.what() <<std::flush;
 345         }
 346
 347         return OC_STACK_KEEP_TRANSACTION;
 348     }
 349
 350     OCStackResult InProcClientWrapper::ListenForDevice(
 351             const std::string& serviceUrl,
 352             const std::string& deviceURI,
 353             OCConnectivityType connectivityType,
 354             FindDeviceCallback& callback,
 355             QualityOfService QoS)
 356     {
 357         if (!callback)
 358         {
 359             return OC_STACK_INVALID_PARAM;
 360         }
 361         OCStackResult result;
 362         ostringstream deviceUri;
 363         deviceUri << serviceUrl << deviceURI;
 364
 365         ClientCallbackContext::DeviceListenContext* context =
 366             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
 367         OCCallbackData cbdata;
 368
 369         cbdata.context = static_cast<void*>(context),
 370         cbdata.cb      = listenDeviceCallback;
 371         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
 372
 373         auto cLock = m_csdkLock.lock();
 374         if (cLock)
 375         {
 376             std::lock_guard<std::recursive_mutex> lock(*cLock);
 377             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 378                                   deviceUri.str().c_str(),
 379                                   nullptr, nullptr, connectivityType,
 380                                   static_cast<OCQualityOfService>(QoS),
 381                                   &cbdata,
 382                                   nullptr, 0);
 383         }
 384         else
 385         {
 386             delete context;
 387             result = OC_STACK_ERROR;
 388         }
 389         return result;
 390     }
 391
 392     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 393                     HeaderOptions& serverHeaderOptions)
 394     {
 395         if (clientResponse)
 396         {
 397             // Parse header options from server
 398             uint16_t optionID;
 399             std::string optionData;
 400
 401             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 402             {
 403                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 404                 optionData = reinterpret_cast<const char*>
 405                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 406                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 407                 serverHeaderOptions.push_back(headerOption);
 408             }
 409         }
 410         else
 411         {
 412             // clientResponse is invalid
 413             // TODO check proper logging
 414             std::cout << " Invalid response " << std::endl;
 415         }
 416     }
 417
 418     OCStackApplicationResult getResourceCallback(void* ctx,
 419                                                  OCDoHandle /*handle*/,
 420         OCClientResponse* clientResponse)
 421     {
 422         ClientCallbackContext::GetContext* context =
 423             static_cast<ClientCallbackContext::GetContext*>(ctx);
 424
 425         OCRepresentation rep;
 426         HeaderOptions serverHeaderOptions;
 427         OCStackResult result = clientResponse->result;
 428         if (result == OC_STACK_OK)
 429         {
 430             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 431             try
 432             {
 433                 rep = parseGetSetCallback(clientResponse);
 434             }
 435             catch(OC::OCException& e)
 436             {
 437                 result = e.code();
 438             }
 439         }
 440
 441         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 442         exec.detach();
 443         return OC_STACK_DELETE_TRANSACTION;
 444     }
 445
 446     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 447         const OCDevAddr& devAddr,
 448         const std::string& resourceUri,
 449         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 450         GetCallback& callback, QualityOfService QoS)
 451     {
 452         if (!callback)
 453         {
 454             return OC_STACK_INVALID_PARAM;
 455         }
 456         OCStackResult result;
 457         ClientCallbackContext::GetContext* ctx =
 458             new ClientCallbackContext::GetContext(callback);
 459         OCCallbackData cbdata;
 460         cbdata.context = static_cast<void*>(ctx),
 461         cbdata.cb      = getResourceCallback;
 462         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 463
 464
 465         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 466
 467         auto cLock = m_csdkLock.lock();
 468
 469         if (cLock)
 470         {
 471             std::lock_guard<std::recursive_mutex> lock(*cLock);
 472             OCHeaderOption options[MAX_HEADER_OPTIONS];
 473
 474             result = OCDoResource(
 475                                   nullptr, OC_REST_GET,
 476                                   uri.c_str(),
 477                                   &devAddr, nullptr,
 478                                   CT_DEFAULT,
 479                                   static_cast<OCQualityOfService>(QoS),
 480                                   &cbdata,
 481                                   assembleHeaderOptions(options, headerOptions),
 482                                   headerOptions.size());
 483         }
 484         else
 485         {
 486             delete ctx;
 487             result = OC_STACK_ERROR;
 488         }
 489         return result;
 490     }
 491
 492
 493     OCStackApplicationResult setResourceCallback(void* ctx,
 494                                                  OCDoHandle /*handle*/,
 495         OCClientResponse* clientResponse)
 496     {
 497         ClientCallbackContext::SetContext* context =
 498             static_cast<ClientCallbackContext::SetContext*>(ctx);
 499         OCRepresentation attrs;
 500         HeaderOptions serverHeaderOptions;
 501
 502         OCStackResult result = clientResponse->result;
 503         if (OC_STACK_OK               == result ||
 504             OC_STACK_RESOURCE_CREATED == result ||
 505             OC_STACK_RESOURCE_DELETED == result ||
 506             OC_STACK_RESOURCE_CHANGED == result)
 507         {
 508             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 509             try
 510             {
 511                 attrs = parseGetSetCallback(clientResponse);
 512             }
 513             catch(OC::OCException& e)
 514             {
 515                 result = e.code();
 516             }
 517         }
 518
 519         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
 520         exec.detach();
 521         return OC_STACK_DELETE_TRANSACTION;
 522     }
 523
 524     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 525         const QueryParamsMap& queryParams)
 526     {
 527         if (!uri.empty())
 528         {
 529             if (uri.back() == '/')
 530             {
 531                 uri.resize(uri.size() - 1);
 532             }
 533         }
 534
 535         ostringstream paramsList;
 536         if (queryParams.size() > 0)
 537         {
 538             paramsList << '?';
 539         }
 540
 541         for (auto& param : queryParams)
 542         {
 543             paramsList << param.first <<'='<<param.second<<';';
 544         }
 545
 546         std::string queryString = paramsList.str();
 547
 548         if (queryString.empty())
 549         {
 550             return uri;
 551         }
 552
 553         if (queryString.back() == ';')
 554         {
 555             queryString.resize(queryString.size() - 1);
 556         }
 557
 558         std::string ret = uri + queryString;
 559         return ret;
 560     }
 561
 562     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 563     {
 564         MessageContainer ocInfo;
 565         ocInfo.addRepresentation(rep);
 566         for(const OCRepresentation& r : rep.getChildren())
 567         {
 568             ocInfo.addRepresentation(r);
 569         }
 570
 571         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 572     }
 573
 574     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 575         const OCDevAddr& devAddr,
 576         const std::string& uri,
 577         const OCRepresentation& rep,
 578         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 579         PostCallback& callback, QualityOfService QoS)
 580     {
 581         if (!callback)
 582         {
 583             return OC_STACK_INVALID_PARAM;
 584         }
 585         OCStackResult result;
 586         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 587         OCCallbackData cbdata;
 588         cbdata.context = static_cast<void*>(ctx),
 589         cbdata.cb      = setResourceCallback;
 590         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 591
 592
 593         std::string url = assembleSetResourceUri(uri, queryParams);
 594
 595         auto cLock = m_csdkLock.lock();
 596
 597         if (cLock)
 598         {
 599             std::lock_guard<std::recursive_mutex> lock(*cLock);
 600             OCHeaderOption options[MAX_HEADER_OPTIONS];
 601
 602             result = OCDoResource(nullptr, OC_REST_POST,
 603                                   url.c_str(), &devAddr,
 604                                   assembleSetResourcePayload(rep),
 605                                   CT_DEFAULT,
 606                                   static_cast<OCQualityOfService>(QoS),
 607                                   &cbdata,
 608                                   assembleHeaderOptions(options, headerOptions),
 609                                   headerOptions.size());
 610         }
 611         else
 612         {
 613             delete ctx;
 614             result = OC_STACK_ERROR;
 615         }
 616
 617         return result;
 618     }
 619
 620     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 621         const OCDevAddr& devAddr,
 622         const std::string& uri,
 623         const OCRepresentation& rep,
 624         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 625         PutCallback& callback, QualityOfService QoS)
 626     {
 627         if (!callback)
 628         {
 629             return OC_STACK_INVALID_PARAM;
 630         }
 631         OCStackResult result;
 632         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 633         OCCallbackData cbdata;
 634         cbdata.context = static_cast<void*>(ctx),
 635         cbdata.cb      = setResourceCallback;
 636         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 637
 638
 639         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 640
 641         auto cLock = m_csdkLock.lock();
 642
 643         if (cLock)
 644         {
 645             std::lock_guard<std::recursive_mutex> lock(*cLock);
 646             OCDoHandle handle;
 647             OCHeaderOption options[MAX_HEADER_OPTIONS];
 648
 649             result = OCDoResource(&handle, OC_REST_PUT,
 650                                   url.c_str(), &devAddr,
 651                                   assembleSetResourcePayload(rep),
 652                                   CT_DEFAULT,
 653                                   static_cast<OCQualityOfService>(QoS),
 654                                   &cbdata,
 655                                   assembleHeaderOptions(options, headerOptions),
 656                                   headerOptions.size());
 657         }
 658         else
 659         {
 660             delete ctx;
 661             result = OC_STACK_ERROR;
 662         }
 663
 664         return result;
 665     }
 666
 667     OCStackApplicationResult deleteResourceCallback(void* ctx,
 668                                                     OCDoHandle /*handle*/,
 669         OCClientResponse* clientResponse)
 670     {
 671         ClientCallbackContext::DeleteContext* context =
 672             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 673         HeaderOptions serverHeaderOptions;
 674
 675         if (clientResponse->result == OC_STACK_OK)
 676         {
 677             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 678         }
 679         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 680         exec.detach();
 681         return OC_STACK_DELETE_TRANSACTION;
 682     }
 683
 684     OCStackResult InProcClientWrapper::DeleteResource(
 685         const OCDevAddr& devAddr,
 686         const std::string& uri,
 687         const HeaderOptions& headerOptions,
 688         DeleteCallback& callback,
 689         QualityOfService /*QoS*/)
 690     {
 691         if (!callback)
 692         {
 693             return OC_STACK_INVALID_PARAM;
 694         }
 695         OCStackResult result;
 696         ClientCallbackContext::DeleteContext* ctx =
 697             new ClientCallbackContext::DeleteContext(callback);
 698         OCCallbackData cbdata;
 699         cbdata.context = static_cast<void*>(ctx),
 700         cbdata.cb      = deleteResourceCallback;
 701         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
 702
 703
 704         auto cLock = m_csdkLock.lock();
 705
 706         if (cLock)
 707         {
 708             OCHeaderOption options[MAX_HEADER_OPTIONS];
 709
 710             std::lock_guard<std::recursive_mutex> lock(*cLock);
 711
 712             result = OCDoResource(nullptr, OC_REST_DELETE,
 713                                   uri.c_str(), &devAddr,
 714                                   nullptr,
 715                                   CT_DEFAULT,
 716                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 717                                   &cbdata,
 718                                   assembleHeaderOptions(options, headerOptions),
 719                                   headerOptions.size());
 720         }
 721         else
 722         {
 723             delete ctx;
 724             result = OC_STACK_ERROR;
 725         }
 726
 727         return result;
 728     }
 729
 730     OCStackApplicationResult observeResourceCallback(void* ctx,
 731                                                      OCDoHandle /*handle*/,
 732         OCClientResponse* clientResponse)
 733     {
 734         ClientCallbackContext::ObserveContext* context =
 735             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 736         OCRepresentation attrs;
 737         HeaderOptions serverHeaderOptions;
 738         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 739         OCStackResult result = clientResponse->result;
 740         if (clientResponse->result == OC_STACK_OK)
 741         {
 742             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 743             try
 744             {
 745                 attrs = parseGetSetCallback(clientResponse);
 746             }
 747             catch(OC::OCException& e)
 748             {
 749                 result = e.code();
 750             }
 751         }
 752         std::thread exec(context->callback, serverHeaderOptions, attrs,
 753                     result, sequenceNumber);
 754         exec.detach();
 755         if (sequenceNumber == OC_OBSERVE_DEREGISTER)
 756         {
 757             return OC_STACK_DELETE_TRANSACTION;
 758         }
 759         return OC_STACK_KEEP_TRANSACTION;
 760     }
 761
 762     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 763         const OCDevAddr& devAddr,
 764         const std::string& uri,
 765         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 766         ObserveCallback& callback, QualityOfService QoS)
 767     {
 768         if (!callback)
 769         {
 770             return OC_STACK_INVALID_PARAM;
 771         }
 772         OCStackResult result;
 773
 774         ClientCallbackContext::ObserveContext* ctx =
 775             new ClientCallbackContext::ObserveContext(callback);
 776         OCCallbackData cbdata;
 777         cbdata.context = static_cast<void*>(ctx),
 778         cbdata.cb      = observeResourceCallback;
 779         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
 780
 781
 782         OCMethod method;
 783         if (observeType == ObserveType::Observe)
 784         {
 785             method = OC_REST_OBSERVE;
 786         }
 787         else if (observeType == ObserveType::ObserveAll)
 788         {
 789             method = OC_REST_OBSERVE_ALL;
 790         }
 791         else
 792         {
 793             method = OC_REST_OBSERVE_ALL;
 794         }
 795
 796         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 797
 798         auto cLock = m_csdkLock.lock();
 799
 800         if (cLock)
 801         {
 802             std::lock_guard<std::recursive_mutex> lock(*cLock);
 803             OCHeaderOption options[MAX_HEADER_OPTIONS];
 804
 805             result = OCDoResource(handle, method,
 806                                   url.c_str(), &devAddr,
 807                                   nullptr,
 808                                   CT_DEFAULT,
 809                                   static_cast<OCQualityOfService>(QoS),
 810                                   &cbdata,
 811                                   assembleHeaderOptions(options, headerOptions),
 812                                   headerOptions.size());
 813         }
 814         else
 815         {
 816             delete ctx;
 817             return OC_STACK_ERROR;
 818         }
 819
 820         return result;
 821     }
 822
 823     OCStackResult InProcClientWrapper::CancelObserveResource(
 824             OCDoHandle handle,
 825             const std::string& /*host*/,
 826             const std::string& /*uri*/,
 827             const HeaderOptions& headerOptions,
 828             QualityOfService QoS)
 829     {
 830         OCStackResult result;
 831         auto cLock = m_csdkLock.lock();
 832
 833         if (cLock)
 834         {
 835             std::lock_guard<std::recursive_mutex> lock(*cLock);
 836             OCHeaderOption options[MAX_HEADER_OPTIONS];
 837
 838             result = OCCancel(handle,
 839                     static_cast<OCQualityOfService>(QoS),
 840                     assembleHeaderOptions(options, headerOptions),
 841                     headerOptions.size());
 842         }
 843         else
 844         {
 845             result = OC_STACK_ERROR;
 846         }
 847
 848         return result;
 849     }
 850
 851     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 852                                                        OCDoHandle /*handle*/,
 853             OCClientResponse* clientResponse)
 854     {
 855         ClientCallbackContext::SubscribePresenceContext* context =
 856         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 857
 858         /*
 859          * This a hack while we rethink presence subscription.
 860          */
 861         std::string url = clientResponse->devAddr.addr;
 862
 863         std::thread exec(context->callback, clientResponse->result,
 864                     clientResponse->sequenceNumber, url);
 865
 866         exec.detach();
 867
 868         return OC_STACK_KEEP_TRANSACTION;
 869     }
 870
 871     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 872         const std::string& host, const std::string& resourceType,
 873         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 874     {
 875         if (!presenceHandler)
 876         {
 877             return OC_STACK_INVALID_PARAM;
 878         }
 879
 880         ClientCallbackContext::SubscribePresenceContext* ctx =
 881             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 882         OCCallbackData cbdata;
 883         cbdata.context = static_cast<void*>(ctx),
 884         cbdata.cb      = subscribePresenceCallback;
 885         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
 886
 887
 888         auto cLock = m_csdkLock.lock();
 889
 890         std::ostringstream os;
 891         os << host << OC_RSRVD_PRESENCE_URI;
 892
 893         if (!resourceType.empty())
 894         {
 895             os << "?rt=" << resourceType;
 896         }
 897
 898         if (!cLock)
 899         {
 900             delete ctx;
 901             return OC_STACK_ERROR;
 902         }
 903
 904         return OCDoResource(handle, OC_REST_PRESENCE,
 905                             os.str().c_str(), nullptr,
 906                             nullptr, connectivityType,
 907                             OC_LOW_QOS, &cbdata, NULL, 0);
 908     }
 909
 910     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 911     {
 912         OCStackResult result;
 913         auto cLock = m_csdkLock.lock();
 914
 915         if (cLock)
 916         {
 917             std::lock_guard<std::recursive_mutex> lock(*cLock);
 918             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 919         }
 920         else
 921         {
 922             result = OC_STACK_ERROR;
 923         }
 924
 925         return result;
 926     }
 927
 928     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 929     {
 930         qos = m_cfg.QoS;
 931         return OC_STACK_OK;
 932     }
 933
 934     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 935            const HeaderOptions& headerOptions)
 936     {
 937         int i = 0;
 938
 939         if ( headerOptions.size() == 0)
 940         {
 941             return nullptr;
 942         }
 943
 944         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 945         {
 946             options[i] = OCHeaderOption();
 947             options[i].protocolID = OC_COAP_ID;
 948             options[i].optionID = it->getOptionID();
 949             options[i].optionLength = it->getOptionData().length() + 1;
 950             strcpy((char*)options[i].optionData, (it->getOptionData().c_str()));
 951             i++;
 952         }
 953
 954         return options;
 955     }
 956
 957     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
 958     {
 959         if (!dev)
 960         {
 961             return nullptr;
 962         }
 963
 964         OCDPDev_t* result = new OCDPDev_t(*dev);
 965         result->prm = new OCPrm_t[dev->prmLen];
 966         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
 967         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
 968     }
 969
 970     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
 971     {
 972         while(list)
 973         {
 974             dpList.push_back(cloneDevice(list));
 975             list = list->next;
 976         }
 977     }
 978
 979     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
 980             GetDirectPairedCallback& callback)
 981     {
 982         if (!callback || 0 == waittime)
 983         {
 984             return OC_STACK_INVALID_PARAM;
 985         }
 986
 987         OCStackResult result = OC_STACK_ERROR;
 988         const OCDPDev_t *list = nullptr;
 989         PairedDevices dpDeviceList;
 990
 991         auto cLock = m_csdkLock.lock();
 992
 993         if (cLock)
 994         {
 995             std::lock_guard<std::recursive_mutex> lock(*cLock);
 996
 997             list = OCDiscoverDirectPairingDevices(waittime);
 998             if (NULL == list)
 999             {
1000                 result = OC_STACK_NO_RESOURCE;
1001                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1002                     << std::flush;
1003             }
1004             else {
1005                 convert(list, dpDeviceList);
1006                 std::thread exec(callback, dpDeviceList);
1007                 exec.detach();
1008                 result = OC_STACK_OK;
1009             }
1010         }
1011         else
1012         {
1013             result = OC_STACK_ERROR;
1014         }
1015
1016         return result;
1017     }
1018
1019     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1020     {
1021         if (!callback)
1022         {
1023             return OC_STACK_INVALID_PARAM;
1024         }
1025
1026         OCStackResult result = OC_STACK_ERROR;
1027         const OCDPDev_t *list = nullptr;
1028         PairedDevices dpDeviceList;
1029
1030         auto cLock = m_csdkLock.lock();
1031
1032         if (cLock)
1033         {
1034             std::lock_guard<std::recursive_mutex> lock(*cLock);
1035
1036             list = OCGetDirectPairedDevices();
1037             if (NULL == list)
1038             {
1039                 result = OC_STACK_NO_RESOURCE;
1040                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1041                     << std::flush;
1042             }
1043             else {
1044                 convert(list, dpDeviceList);
1045                 std::thread exec(callback, dpDeviceList);
1046                 exec.detach();
1047                 result = OC_STACK_OK;
1048             }
1049         }
1050         else
1051         {
1052             result = OC_STACK_ERROR;
1053         }
1054
1055         return result;
1056     }
1057
1058     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1059             OCStackResult result)
1060     {
1061
1062         ClientCallbackContext::DirectPairingContext* context =
1063             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1064
1065         std::thread exec(context->callback, cloneDevice(peer), result);
1066         exec.detach();
1067     }
1068
1069     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1070             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1071     {
1072         if (!peer || !callback)
1073         {
1074             oclog() << "Invalid parameters" << std::flush;
1075             return OC_STACK_INVALID_PARAM;
1076         }
1077
1078         OCStackResult result = OC_STACK_ERROR;
1079         ClientCallbackContext::DirectPairingContext* context =
1080             new ClientCallbackContext::DirectPairingContext(callback);
1081
1082         auto cLock = m_csdkLock.lock();
1083         if (cLock)
1084         {
1085             std::lock_guard<std::recursive_mutex> lock(*cLock);
1086             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1087                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1088         }
1089         else
1090         {
1091             delete context;
1092             result = OC_STACK_ERROR;
1093         }
1094         return result;
1095     }
1096 }