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 = (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 = (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 = (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         {
 507             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 508             try
 509             {
 510                 attrs = parseGetSetCallback(clientResponse);
 511             }
 512             catch(OC::OCException& e)
 513             {
 514                 result = e.code();
 515             }
 516         }
 517
 518         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
 519         exec.detach();
 520         return OC_STACK_DELETE_TRANSACTION;
 521     }
 522
 523     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
 524         const QueryParamsMap& queryParams)
 525     {
 526         if(!uri.empty())
 527         {
 528             if(uri.back() == '/')
 529             {
 530                 uri.resize(uri.size()-1);
 531             }
 532         }
 533
 534         std::ostringstream paramsList;
 535         if(queryParams.size() > 0)
 536         {
 537             paramsList << '?';
 538         }
 539
 540         for(auto& param : queryParams)
 541         {
 542             paramsList << param.first <<'='<<param.second<<';';
 543         }
 544
 545         std::string queryString = paramsList.str();
 546
 547         if(queryString.empty())
 548         {
 549             return uri;
 550         }
 551
 552         if(queryString.back() == ';')
 553         {
 554             queryString.resize(queryString.size() - 1);
 555         }
 556
 557         std::string ret = uri + queryString;
 558         return ret;
 559     }
 560
 561     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
 562     {
 563         MessageContainer ocInfo;
 564         ocInfo.addRepresentation(rep);
 565         for(const OCRepresentation& r : rep.getChildren())
 566         {
 567             ocInfo.addRepresentation(r);
 568         }
 569
 570         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
 571     }
 572
 573     OCStackResult InProcClientWrapper::PostResourceRepresentation(
 574         const OCDevAddr& devAddr,
 575         const std::string& uri,
 576         const OCRepresentation& rep,
 577         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 578         PostCallback& callback, QualityOfService QoS)
 579     {
 580         if(!callback)
 581         {
 582             return OC_STACK_INVALID_PARAM;
 583         }
 584         OCStackResult result;
 585         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 586         OCCallbackData cbdata;
 587         cbdata.context = (void*)ctx;
 588         cbdata.cb      = setResourceCallback;
 589         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 590
 591
 592         std::string url = assembleSetResourceUri(uri, queryParams);
 593
 594         auto cLock = m_csdkLock.lock();
 595
 596         if(cLock)
 597         {
 598             std::lock_guard<std::recursive_mutex> lock(*cLock);
 599             OCHeaderOption options[MAX_HEADER_OPTIONS];
 600
 601             result = OCDoResource(nullptr, OC_REST_POST,
 602                                   url.c_str(), &devAddr,
 603                                   assembleSetResourcePayload(rep),
 604                                   CT_DEFAULT,
 605                                   static_cast<OCQualityOfService>(QoS),
 606                                   &cbdata,
 607                                   assembleHeaderOptions(options, headerOptions),
 608                                   headerOptions.size());
 609         }
 610         else
 611         {
 612             delete ctx;
 613             result = OC_STACK_ERROR;
 614         }
 615
 616         return result;
 617     }
 618
 619     OCStackResult InProcClientWrapper::PutResourceRepresentation(
 620         const OCDevAddr& devAddr,
 621         const std::string& uri,
 622         const OCRepresentation& rep,
 623         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 624         PutCallback& callback, QualityOfService QoS)
 625     {
 626         if(!callback)
 627         {
 628             return OC_STACK_INVALID_PARAM;
 629         }
 630         OCStackResult result;
 631         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
 632         OCCallbackData cbdata;
 633         cbdata.context = (void*)ctx;
 634         cbdata.cb      = setResourceCallback;
 635         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 636
 637
 638         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 639
 640         auto cLock = m_csdkLock.lock();
 641
 642         if(cLock)
 643         {
 644             std::lock_guard<std::recursive_mutex> lock(*cLock);
 645             OCDoHandle handle;
 646             OCHeaderOption options[MAX_HEADER_OPTIONS];
 647
 648             result = OCDoResource(&handle, OC_REST_PUT,
 649                                   url.c_str(), &devAddr,
 650                                   assembleSetResourcePayload(rep),
 651                                   CT_DEFAULT,
 652                                   static_cast<OCQualityOfService>(QoS),
 653                                   &cbdata,
 654                                   assembleHeaderOptions(options, headerOptions),
 655                                   headerOptions.size());
 656         }
 657         else
 658         {
 659             delete ctx;
 660             result = OC_STACK_ERROR;
 661         }
 662
 663         return result;
 664     }
 665
 666     OCStackApplicationResult deleteResourceCallback(void* ctx,
 667                                                     OCDoHandle /*handle*/,
 668         OCClientResponse* clientResponse)
 669     {
 670         ClientCallbackContext::DeleteContext* context =
 671             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
 672         HeaderOptions serverHeaderOptions;
 673
 674         if(clientResponse->result == OC_STACK_OK)
 675         {
 676             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 677         }
 678         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
 679         exec.detach();
 680         return OC_STACK_DELETE_TRANSACTION;
 681     }
 682
 683     OCStackResult InProcClientWrapper::DeleteResource(
 684         const OCDevAddr& devAddr,
 685         const std::string& uri,
 686         const HeaderOptions& headerOptions,
 687         DeleteCallback& callback,
 688         QualityOfService /*QoS*/)
 689     {
 690         if(!callback)
 691         {
 692             return OC_STACK_INVALID_PARAM;
 693         }
 694         OCStackResult result;
 695         ClientCallbackContext::DeleteContext* ctx =
 696             new ClientCallbackContext::DeleteContext(callback);
 697         OCCallbackData cbdata;
 698         cbdata.context = (void*)ctx;
 699         cbdata.cb      = deleteResourceCallback;
 700         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
 701
 702
 703         auto cLock = m_csdkLock.lock();
 704
 705         if(cLock)
 706         {
 707             OCHeaderOption options[MAX_HEADER_OPTIONS];
 708
 709             std::lock_guard<std::recursive_mutex> lock(*cLock);
 710
 711             result = OCDoResource(nullptr, OC_REST_DELETE,
 712                                   uri.c_str(), &devAddr,
 713                                   nullptr,
 714                                   CT_DEFAULT,
 715                                   static_cast<OCQualityOfService>(m_cfg.QoS),
 716                                   &cbdata,
 717                                   assembleHeaderOptions(options, headerOptions),
 718                                   headerOptions.size());
 719         }
 720         else
 721         {
 722             delete ctx;
 723             result = OC_STACK_ERROR;
 724         }
 725
 726         return result;
 727     }
 728
 729     OCStackApplicationResult observeResourceCallback(void* ctx,
 730                                                      OCDoHandle /*handle*/,
 731         OCClientResponse* clientResponse)
 732     {
 733         ClientCallbackContext::ObserveContext* context =
 734             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
 735         OCRepresentation attrs;
 736         HeaderOptions serverHeaderOptions;
 737         uint32_t sequenceNumber = clientResponse->sequenceNumber;
 738         OCStackResult result = clientResponse->result;
 739         if(clientResponse->result == OC_STACK_OK)
 740         {
 741             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 742             try
 743             {
 744                 attrs = parseGetSetCallback(clientResponse);
 745             }
 746             catch(OC::OCException& e)
 747             {
 748                 result = e.code();
 749             }
 750         }
 751         std::thread exec(context->callback, serverHeaderOptions, attrs,
 752                     result, sequenceNumber);
 753         exec.detach();
 754         if(sequenceNumber == OC_OBSERVE_DEREGISTER)
 755         {
 756             return OC_STACK_DELETE_TRANSACTION;
 757         }
 758         return OC_STACK_KEEP_TRANSACTION;
 759     }
 760
 761     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
 762         const OCDevAddr& devAddr,
 763         const std::string& uri,
 764         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 765         ObserveCallback& callback, QualityOfService QoS)
 766     {
 767         if(!callback)
 768         {
 769             return OC_STACK_INVALID_PARAM;
 770         }
 771         OCStackResult result;
 772
 773         ClientCallbackContext::ObserveContext* ctx =
 774             new ClientCallbackContext::ObserveContext(callback);
 775         OCCallbackData cbdata;
 776         cbdata.context = (void*)ctx;
 777         cbdata.cb      = observeResourceCallback;
 778         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
 779
 780
 781         OCMethod method;
 782         if (observeType == ObserveType::Observe)
 783         {
 784             method = OC_REST_OBSERVE;
 785         }
 786         else if (observeType == ObserveType::ObserveAll)
 787         {
 788             method = OC_REST_OBSERVE_ALL;
 789         }
 790         else
 791         {
 792             method = OC_REST_OBSERVE_ALL;
 793         }
 794
 795         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
 796
 797         auto cLock = m_csdkLock.lock();
 798
 799         if(cLock)
 800         {
 801             std::lock_guard<std::recursive_mutex> lock(*cLock);
 802             OCHeaderOption options[MAX_HEADER_OPTIONS];
 803
 804             result = OCDoResource(handle, method,
 805                                   url.c_str(), &devAddr,
 806                                   nullptr,
 807                                   CT_DEFAULT,
 808                                   static_cast<OCQualityOfService>(QoS),
 809                                   &cbdata,
 810                                   assembleHeaderOptions(options, headerOptions),
 811                                   headerOptions.size());
 812         }
 813         else
 814         {
 815             delete ctx;
 816             return OC_STACK_ERROR;
 817         }
 818
 819         return result;
 820     }
 821
 822     OCStackResult InProcClientWrapper::CancelObserveResource(
 823             OCDoHandle handle,
 824             const std::string& /*host*/,
 825             const std::string& /*uri*/,
 826             const HeaderOptions& headerOptions,
 827             QualityOfService QoS)
 828     {
 829         OCStackResult result;
 830         auto cLock = m_csdkLock.lock();
 831
 832         if(cLock)
 833         {
 834             std::lock_guard<std::recursive_mutex> lock(*cLock);
 835             OCHeaderOption options[MAX_HEADER_OPTIONS];
 836
 837             result = OCCancel(handle,
 838                     static_cast<OCQualityOfService>(QoS),
 839                     assembleHeaderOptions(options, headerOptions),
 840                     headerOptions.size());
 841         }
 842         else
 843         {
 844             result = OC_STACK_ERROR;
 845         }
 846
 847         return result;
 848     }
 849
 850     OCStackApplicationResult subscribePresenceCallback(void* ctx,
 851                                                        OCDoHandle /*handle*/,
 852             OCClientResponse* clientResponse)
 853     {
 854         ClientCallbackContext::SubscribePresenceContext* context =
 855         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
 856
 857         /*
 858          * This a hack while we rethink presence subscription.
 859          */
 860         std::string url = clientResponse->devAddr.addr;
 861
 862         std::thread exec(context->callback, clientResponse->result,
 863                     clientResponse->sequenceNumber, url);
 864
 865         exec.detach();
 866
 867         return OC_STACK_KEEP_TRANSACTION;
 868     }
 869
 870     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
 871         const std::string& host, const std::string& resourceType,
 872         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
 873     {
 874         if(!presenceHandler)
 875         {
 876             return OC_STACK_INVALID_PARAM;
 877         }
 878
 879         ClientCallbackContext::SubscribePresenceContext* ctx =
 880             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
 881         OCCallbackData cbdata;
 882         cbdata.context = (void*)ctx;
 883         cbdata.cb      = subscribePresenceCallback;
 884         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
 885
 886
 887         auto cLock = m_csdkLock.lock();
 888
 889         std::ostringstream os;
 890         os << host << OC_RSRVD_PRESENCE_URI;
 891
 892         if(!resourceType.empty())
 893         {
 894             os << "?rt=" << resourceType;
 895         }
 896
 897         if(!cLock)
 898         {
 899             delete ctx;
 900             return OC_STACK_ERROR;
 901         }
 902
 903         return OCDoResource(handle, OC_REST_PRESENCE,
 904                             os.str().c_str(), nullptr,
 905                             nullptr, connectivityType,
 906                             OC_LOW_QOS, &cbdata, NULL, 0);
 907     }
 908
 909     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
 910     {
 911         OCStackResult result;
 912         auto cLock = m_csdkLock.lock();
 913
 914         if(cLock)
 915         {
 916             std::lock_guard<std::recursive_mutex> lock(*cLock);
 917             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
 918         }
 919         else
 920         {
 921             result = OC_STACK_ERROR;
 922         }
 923
 924         return result;
 925     }
 926
 927     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
 928     {
 929         qos = m_cfg.QoS;
 930         return OC_STACK_OK;
 931     }
 932
 933     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
 934            const HeaderOptions& headerOptions)
 935     {
 936         int i = 0;
 937
 938         if( headerOptions.size() == 0)
 939         {
 940             return nullptr;
 941         }
 942
 943         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
 944         {
 945             options[i] = OCHeaderOption();
 946             options[i].protocolID = OC_COAP_ID;
 947             options[i].optionID = it->getOptionID();
 948             options[i].optionLength = it->getOptionData().length() + 1;
 949             strcpy((char*)options[i].optionData, (it->getOptionData().c_str()));
 950             i++;
 951         }
 952
 953         return options;
 954     }
 955 }