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         OCConnectivityType connectivityType,
 580         PostCallback& callback, QualityOfService QoS)
 581     {
 582         if (!callback)
 583         {
 584             return OC_STACK_INVALID_PARAM;
 585         }
 586         OCStackResult result;
 587         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 588         OCCallbackData cbdata;
 589         cbdata.context = static_cast<void*>(ctx),
 590         cbdata.cb      = setResourceCallback;
 591         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 592
 593
 594         std::string url = assembleSetResourceUri(uri, queryParams);
 595
 596         auto cLock = m_csdkLock.lock();
 597
 598         if (cLock)
 599         {
 600             std::lock_guard<std::recursive_mutex> lock(*cLock);
 601             OCHeaderOption options[MAX_HEADER_OPTIONS];
 602
 603             result = OCDoResource(nullptr, OC_REST_POST,
 604                                   url.c_str(), &devAddr,
 605                                   assembleSetResourcePayload(rep),
 606                                   connectivityType,
 607                                   static_cast<OCQualityOfService>(QoS),
 608                                   &cbdata,
 609                                   assembleHeaderOptions(options, headerOptions),
 610                                   headerOptions.size());
 611         }
 612         else
 613         {
 614             delete ctx;
 615             result = OC_STACK_ERROR;
 616         }
 617
 618         return result;
 619     }
 620
 621     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 622         const OCDevAddr& devAddr,
 623         const std::string& uri,
 624         const OCRepresentation& rep,
 625         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 626         PutCallback& callback, QualityOfService QoS)
 627     {
 628         if (!callback)
 629         {
 630             return OC_STACK_INVALID_PARAM;
 631         }
 632         OCStackResult result;
 633         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 634         OCCallbackData cbdata;
 635         cbdata.context = static_cast<void*>(ctx),
 636         cbdata.cb      = setResourceCallback;
 637         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 638
 639
 640         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 641
 642         auto cLock = m_csdkLock.lock();
 643
 644         if (cLock)
 645         {
 646             std::lock_guard<std::recursive_mutex> lock(*cLock);
 647             OCDoHandle handle;
 648             OCHeaderOption options[MAX_HEADER_OPTIONS];
 649
 650             result = OCDoResource(&handle, OC_REST_PUT,
 651                                   url.c_str(), &devAddr,
 652                                   assembleSetResourcePayload(rep),
 653                                   CT_DEFAULT,
 654                                   static_cast<OCQualityOfService>(QoS),
 655                                   &cbdata,
 656                                   assembleHeaderOptions(options, headerOptions),
 657                                   headerOptions.size());
 658         }
 659         else
 660         {
 661             delete ctx;
 662             result = OC_STACK_ERROR;
 663         }
 664
 665         return result;
 666     }
 667
 668     OCStackApplicationResult deleteResourceCallback(void* ctx,
 669                                                     OCDoHandle /*handle*/,
 670         OCClientResponse* clientResponse)
 671     {
 672         ClientCallbackContext::DeleteContext* context =
 673             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 674         HeaderOptions serverHeaderOptions;
 675
 676         if (clientResponse->result == OC_STACK_OK)
 677         {
 678             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 679         }
 680         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 681         exec.detach();
 682         return OC_STACK_DELETE_TRANSACTION;
 683     }
 684
 685     OCStackResult InProcClientWrapper::DeleteResource(
 686         const OCDevAddr& devAddr,
 687         const std::string& uri,
 688         const HeaderOptions& headerOptions,
 689         DeleteCallback& callback,
 690         QualityOfService /*QoS*/)
 691     {
 692         if (!callback)
 693         {
 694             return OC_STACK_INVALID_PARAM;
 695         }
 696         OCStackResult result;
 697         ClientCallbackContext::DeleteContext* ctx =
 698             new ClientCallbackContext::DeleteContext(callback);
 699         OCCallbackData cbdata;
 700         cbdata.context = static_cast<void*>(ctx),
 701         cbdata.cb      = deleteResourceCallback;
 702         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
 703
 704
 705         auto cLock = m_csdkLock.lock();
 706
 707         if (cLock)
 708         {
 709             OCHeaderOption options[MAX_HEADER_OPTIONS];
 710
 711             std::lock_guard<std::recursive_mutex> lock(*cLock);
 712
 713             result = OCDoResource(nullptr, OC_REST_DELETE,
 714                                   uri.c_str(), &devAddr,
 715                                   nullptr,
 716                                   CT_DEFAULT,
 717                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 718                                   &cbdata,
 719                                   assembleHeaderOptions(options, headerOptions),
 720                                   headerOptions.size());
 721         }
 722         else
 723         {
 724             delete ctx;
 725             result = OC_STACK_ERROR;
 726         }
 727
 728         return result;
 729     }
 730
 731     OCStackApplicationResult observeResourceCallback(void* ctx,
 732                                                      OCDoHandle /*handle*/,
 733         OCClientResponse* clientResponse)
 734     {
 735         ClientCallbackContext::ObserveContext* context =
 736             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 737         OCRepresentation attrs;
 738         HeaderOptions serverHeaderOptions;
 739         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 740         OCStackResult result = clientResponse->result;
 741         if (clientResponse->result == OC_STACK_OK)
 742         {
 743             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 744             try
 745             {
 746                 attrs = parseGetSetCallback(clientResponse);
 747             }
 748             catch(OC::OCException& e)
 749             {
 750                 result = e.code();
 751             }
 752         }
 753         std::thread exec(context->callback, serverHeaderOptions, attrs,
 754                     result, sequenceNumber);
 755         exec.detach();
 756         if (sequenceNumber == OC_OBSERVE_DEREGISTER)
 757         {
 758             return OC_STACK_DELETE_TRANSACTION;
 759         }
 760         return OC_STACK_KEEP_TRANSACTION;
 761     }
 762
 763     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 764         const OCDevAddr& devAddr,
 765         const std::string& uri,
 766         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 767         ObserveCallback& callback, QualityOfService QoS)
 768     {
 769         if (!callback)
 770         {
 771             return OC_STACK_INVALID_PARAM;
 772         }
 773         OCStackResult result;
 774
 775         ClientCallbackContext::ObserveContext* ctx =
 776             new ClientCallbackContext::ObserveContext(callback);
 777         OCCallbackData cbdata;
 778         cbdata.context = static_cast<void*>(ctx),
 779         cbdata.cb      = observeResourceCallback;
 780         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
 781
 782
 783         OCMethod method;
 784         if (observeType == ObserveType::Observe)
 785         {
 786             method = OC_REST_OBSERVE;
 787         }
 788         else if (observeType == ObserveType::ObserveAll)
 789         {
 790             method = OC_REST_OBSERVE_ALL;
 791         }
 792         else
 793         {
 794             method = OC_REST_OBSERVE_ALL;
 795         }
 796
 797         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 798
 799         auto cLock = m_csdkLock.lock();
 800
 801         if (cLock)
 802         {
 803             std::lock_guard<std::recursive_mutex> lock(*cLock);
 804             OCHeaderOption options[MAX_HEADER_OPTIONS];
 805
 806             result = OCDoResource(handle, method,
 807                                   url.c_str(), &devAddr,
 808                                   nullptr,
 809                                   CT_DEFAULT,
 810                                   static_cast<OCQualityOfService>(QoS),
 811                                   &cbdata,
 812                                   assembleHeaderOptions(options, headerOptions),
 813                                   headerOptions.size());
 814         }
 815         else
 816         {
 817             delete ctx;
 818             return OC_STACK_ERROR;
 819         }
 820
 821         return result;
 822     }
 823
 824     OCStackResult InProcClientWrapper::CancelObserveResource(
 825             OCDoHandle handle,
 826             const std::string& /*host*/,
 827             const std::string& /*uri*/,
 828             const HeaderOptions& headerOptions,
 829             QualityOfService QoS)
 830     {
 831         OCStackResult result;
 832         auto cLock = m_csdkLock.lock();
 833
 834         if (cLock)
 835         {
 836             std::lock_guard<std::recursive_mutex> lock(*cLock);
 837             OCHeaderOption options[MAX_HEADER_OPTIONS];
 838
 839             result = OCCancel(handle,
 840                     static_cast<OCQualityOfService>(QoS),
 841                     assembleHeaderOptions(options, headerOptions),
 842                     headerOptions.size());
 843         }
 844         else
 845         {
 846             result = OC_STACK_ERROR;
 847         }
 848
 849         return result;
 850     }
 851
 852     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 853                                                        OCDoHandle /*handle*/,
 854             OCClientResponse* clientResponse)
 855     {
 856         ClientCallbackContext::SubscribePresenceContext* context =
 857         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 858
 859         /*
 860          * This a hack while we rethink presence subscription.
 861          */
 862         std::string url = clientResponse->devAddr.addr;
 863
 864         std::thread exec(context->callback, clientResponse->result,
 865                     clientResponse->sequenceNumber, url);
 866
 867         exec.detach();
 868
 869         return OC_STACK_KEEP_TRANSACTION;
 870     }
 871
 872     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 873         const std::string& host, const std::string& resourceType,
 874         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 875     {
 876         if (!presenceHandler)
 877         {
 878             return OC_STACK_INVALID_PARAM;
 879         }
 880
 881         ClientCallbackContext::SubscribePresenceContext* ctx =
 882             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 883         OCCallbackData cbdata;
 884         cbdata.context = static_cast<void*>(ctx),
 885         cbdata.cb      = subscribePresenceCallback;
 886         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
 887
 888
 889         auto cLock = m_csdkLock.lock();
 890
 891         std::ostringstream os;
 892         os << host << OC_RSRVD_PRESENCE_URI;
 893
 894         if (!resourceType.empty())
 895         {
 896             os << "?rt=" << resourceType;
 897         }
 898
 899         if (!cLock)
 900         {
 901             delete ctx;
 902             return OC_STACK_ERROR;
 903         }
 904
 905         return OCDoResource(handle, OC_REST_PRESENCE,
 906                             os.str().c_str(), nullptr,
 907                             nullptr, connectivityType,
 908                             OC_LOW_QOS, &cbdata, NULL, 0);
 909     }
 910
 911     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 912     {
 913         OCStackResult result;
 914         auto cLock = m_csdkLock.lock();
 915
 916         if (cLock)
 917         {
 918             std::lock_guard<std::recursive_mutex> lock(*cLock);
 919             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 920         }
 921         else
 922         {
 923             result = OC_STACK_ERROR;
 924         }
 925
 926         return result;
 927     }
 928
 929     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 930     {
 931         qos = m_cfg.QoS;
 932         return OC_STACK_OK;
 933     }
 934
 935     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 936            const HeaderOptions& headerOptions)
 937     {
 938         int i = 0;
 939
 940         if ( headerOptions.size() == 0)
 941         {
 942             return nullptr;
 943         }
 944
 945         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 946         {
 947             options[i] = OCHeaderOption();
 948             options[i].protocolID = OC_COAP_ID;
 949             options[i].optionID = it->getOptionID();
 950             options[i].optionLength = it->getOptionData().length() + 1;
 951             strcpy((char*)options[i].optionData, (it->getOptionData().c_str()));
 952             i++;
 953         }
 954
 955         return options;
 956     }
 957
 958     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
 959     {
 960         if (!dev)
 961         {
 962             return nullptr;
 963         }
 964
 965         OCDPDev_t* result = new OCDPDev_t(*dev);
 966         result->prm = new OCPrm_t[dev->prmLen];
 967         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
 968         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
 969     }
 970
 971     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
 972     {
 973         while(list)
 974         {
 975             dpList.push_back(cloneDevice(list));
 976             list = list->next;
 977         }
 978     }
 979
 980     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
 981             GetDirectPairedCallback& callback)
 982     {
 983         if (!callback || 0 == waittime)
 984         {
 985             return OC_STACK_INVALID_PARAM;
 986         }
 987
 988         OCStackResult result = OC_STACK_ERROR;
 989         const OCDPDev_t *list = nullptr;
 990         PairedDevices dpDeviceList;
 991
 992         auto cLock = m_csdkLock.lock();
 993
 994         if (cLock)
 995         {
 996             std::lock_guard<std::recursive_mutex> lock(*cLock);
 997
 998             list = OCDiscoverDirectPairingDevices(waittime);
 999             if (NULL == list)
1000             {
1001                 result = OC_STACK_NO_RESOURCE;
1002                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1003                     << std::flush;
1004             }
1005             else {
1006                 convert(list, dpDeviceList);
1007                 std::thread exec(callback, dpDeviceList);
1008                 exec.detach();
1009                 result = OC_STACK_OK;
1010             }
1011         }
1012         else
1013         {
1014             result = OC_STACK_ERROR;
1015         }
1016
1017         return result;
1018     }
1019
1020     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1021     {
1022         if (!callback)
1023         {
1024             return OC_STACK_INVALID_PARAM;
1025         }
1026
1027         OCStackResult result = OC_STACK_ERROR;
1028         const OCDPDev_t *list = nullptr;
1029         PairedDevices dpDeviceList;
1030
1031         auto cLock = m_csdkLock.lock();
1032
1033         if (cLock)
1034         {
1035             std::lock_guard<std::recursive_mutex> lock(*cLock);
1036
1037             list = OCGetDirectPairedDevices();
1038             if (NULL == list)
1039             {
1040                 result = OC_STACK_NO_RESOURCE;
1041                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1042                     << std::flush;
1043             }
1044             else {
1045                 convert(list, dpDeviceList);
1046                 std::thread exec(callback, dpDeviceList);
1047                 exec.detach();
1048                 result = OC_STACK_OK;
1049             }
1050         }
1051         else
1052         {
1053             result = OC_STACK_ERROR;
1054         }
1055
1056         return result;
1057     }
1058
1059     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1060             OCStackResult result)
1061     {
1062
1063         ClientCallbackContext::DirectPairingContext* context =
1064             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1065
1066         std::thread exec(context->callback, cloneDevice(peer), result);
1067         exec.detach();
1068     }
1069
1070     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1071             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1072     {
1073         if (!peer || !callback)
1074         {
1075             oclog() << "Invalid parameters" << std::flush;
1076             return OC_STACK_INVALID_PARAM;
1077         }
1078
1079         OCStackResult result = OC_STACK_ERROR;
1080         ClientCallbackContext::DirectPairingContext* context =
1081             new ClientCallbackContext::DirectPairingContext(callback);
1082
1083         auto cLock = m_csdkLock.lock();
1084         if (cLock)
1085         {
1086             std::lock_guard<std::recursive_mutex> lock(*cLock);
1087             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1088                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1089         }
1090         else
1091         {
1092             delete context;
1093             result = OC_STACK_ERROR;
1094         }
1095         return result;
1096     }
1097 }