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