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