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 #include "logger.h"
  29 #ifdef TCP_ADAPTER
  30 #include "oickeepalive.h"
  31 #endif
  32
  33 #define TAG "OIC_CLIENT_WRAPPER"
  34
  35 using namespace std;
  36
  37 namespace OC
  38 {
  39     InProcClientWrapper::InProcClientWrapper(
  40         std::weak_ptr<std::recursive_mutex> csdkLock, PlatformConfig cfg)
  41             : m_threadRun(false), m_csdkLock(csdkLock),
  42               m_cfg { cfg }
  43     {
  44         // if the config type is server, we ought to never get called.  If the config type
  45         // is both, we count on the server to run the thread and do the initialize
  46         start();
  47     }
  48
  49     InProcClientWrapper::~InProcClientWrapper()
  50     {
  51         try
  52         {
  53             stop();
  54         }
  55         catch (InitializeException &e)
  56         {
  57             oclog() << "Exception in stop"<< e.what() << std::flush;
  58         }
  59     }
  60
  61     OCStackResult InProcClientWrapper::start()
  62     {
  63         OIC_LOG_V(INFO, TAG, "start ocplatform for client : %d", m_cfg.transportType);
  64
  65         if (m_cfg.mode == ModeType::Client)
  66         {
  67             OCTransportFlags serverFlags =
  68                             static_cast<OCTransportFlags>(m_cfg.serverConnectivity & CT_MASK_FLAGS);
  69             OCTransportFlags clientFlags =
  70                             static_cast<OCTransportFlags>(m_cfg.clientConnectivity & CT_MASK_FLAGS);
  71             OCStackResult result = OCInit2(OC_CLIENT, serverFlags, clientFlags,
  72                                            m_cfg.transportType);
  73
  74             if (OC_STACK_OK != result)
  75             {
  76                 throw InitializeException(OC::InitException::STACK_INIT_ERROR, result);
  77             }
  78
  79             if (false == m_threadRun)
  80             {
  81                 m_threadRun = true;
  82                 m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
  83             }
  84         }
  85         return OC_STACK_OK;
  86     }
  87
  88     OCStackResult InProcClientWrapper::stop()
  89     {
  90         OIC_LOG(INFO, TAG, "stop ocplatform");
  91
  92         if (m_threadRun && m_listeningThread.joinable())
  93         {
  94             m_threadRun = false;
  95             m_listeningThread.join();
  96         }
  97
  98         // only stop if we are the ones who actually called 'start'.  We are counting
  99         // on the server to do the stop.
 100         if (m_cfg.mode == ModeType::Client)
 101         {
 102             OCStackResult result = OCStop();
 103
 104             if (OC_STACK_OK != result)
 105             {
 106                throw InitializeException(OC::InitException::STACK_TERMINATE_ERROR, result);
 107             }
 108         }
 109         return OC_STACK_OK;
 110     }
 111
 112     void InProcClientWrapper::listeningFunc()
 113     {
 114         while(m_threadRun)
 115         {
 116             OCStackResult result;
 117             auto cLock = m_csdkLock.lock();
 118             if (cLock)
 119             {
 120                 std::lock_guard<std::recursive_mutex> lock(*cLock);
 121                 result = OCProcess();
 122             }
 123             else
 124             {
 125                 result = OC_STACK_ERROR;
 126             }
 127
 128             if (result != OC_STACK_OK)
 129             {
 130                 // TODO: do something with result if failed?
 131             }
 132
 133             // To minimize CPU utilization we may wish to do this with sleep
 134             std::this_thread::sleep_for(std::chrono::milliseconds(10));
 135         }
 136     }
 137
 138     OCRepresentation parseGetSetCallback(OCClientResponse* clientResponse)
 139     {
 140         if (clientResponse->payload == nullptr ||
 141                 (
 142                     clientResponse->payload->type != PAYLOAD_TYPE_REPRESENTATION
 143                 )
 144           )
 145         {
 146             return OCRepresentation();
 147         }
 148
 149         MessageContainer oc;
 150         oc.setPayload(clientResponse->payload);
 151
 152         std::vector<OCRepresentation>::const_iterator it = oc.representations().begin();
 153         if (it == oc.representations().end())
 154         {
 155             return OCRepresentation();
 156         }
 157
 158         // first one is considered the root, everything else is considered a child of this one.
 159        OCRepresentation root = *it;
 160        root.setDevAddr(clientResponse->devAddr);
 161        root.setUri(clientResponse->resourceUri);
 162        ++it;
 163
 164         std::for_each(it, oc.representations().end(),
 165                 [&root](const OCRepresentation& repItr)
 166                 {root.addChild(repItr);});
 167         return root;
 168     }
 169
 170     OCStackApplicationResult listenCallback(void* ctx, OCDoHandle /*handle*/,
 171         OCClientResponse* clientResponse)
 172     {
 173         if (!ctx || !clientResponse)
 174         {
 175             return OC_STACK_KEEP_TRANSACTION;
 176         }
 177
 178         ClientCallbackContext::ListenContext* context =
 179             static_cast<ClientCallbackContext::ListenContext*>(ctx);
 180
 181         if (clientResponse->result != OC_STACK_OK)
 182         {
 183             oclog() << "listenCallback(): failed to create resource. clientResponse: "
 184                     << clientResponse->result
 185                     << std::flush;
 186
 187             return OC_STACK_KEEP_TRANSACTION;
 188         }
 189
 190         if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 191         {
 192             oclog() << "listenCallback(): clientResponse payload was null or the wrong type"
 193                 << std::flush;
 194             return OC_STACK_KEEP_TRANSACTION;
 195         }
 196
 197         auto clientWrapper = context->clientWrapper.lock();
 198
 199         if (!clientWrapper)
 200         {
 201             oclog() << "listenCallback(): failed to get a shared_ptr to the client wrapper"
 202                     << std::flush;
 203             return OC_STACK_KEEP_TRANSACTION;
 204         }
 205
 206         try
 207         {
 208             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 209                                     reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 210             // loop to ensure valid construction of all resources
 211
 212             for(auto resource : container.Resources())
 213             {
 214                 std::thread exec(context->callback, resource);
 215                 exec.detach();
 216             }
 217         }
 218         catch (std::exception &e)
 219         {
 220             oclog() << "Exception in listCallback, ignoring response: "
 221                     << e.what() << std::flush;
 222         }
 223
 224
 225         return OC_STACK_KEEP_TRANSACTION;
 226     }
 227
 228     OCStackApplicationResult listenErrorCallback(void* ctx, OCDoHandle /*handle*/,
 229         OCClientResponse* clientResponse)
 230     {
 231         if (!ctx || !clientResponse)
 232         {
 233             return OC_STACK_KEEP_TRANSACTION;
 234         }
 235
 236         ClientCallbackContext::ListenErrorContext* context =
 237             static_cast<ClientCallbackContext::ListenErrorContext*>(ctx);
 238
 239         OCStackResult result = clientResponse->result;
 240         if (result == OC_STACK_OK)
 241         {
 242             if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 243             {
 244                 OIC_LOG_V(DEBUG, TAG, "%s: clientResponse payload was null or the wrong type",
 245                           __func__);
 246                 return OC_STACK_KEEP_TRANSACTION;
 247             }
 248
 249             auto clientWrapper = context->clientWrapper.lock();
 250
 251             if (!clientWrapper)
 252             {
 253                 OIC_LOG_V(DEBUG, TAG, "%s: failed to get a shared_ptr to the client wrapper",
 254                           __func__);
 255                 return OC_STACK_KEEP_TRANSACTION;
 256             }
 257
 258             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 259                                         reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 260             // loop to ensure valid construction of all resources
 261             for (auto resource : container.Resources())
 262             {
 263                 std::thread exec(context->callback, resource);
 264                 exec.detach();
 265             }
 266             return OC_STACK_KEEP_TRANSACTION;
 267         }
 268
 269         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 270         std::string resourceURI;
 271         if(NULL != clientResponse->resourceUri)
 272         {
 273             resourceURI = clientResponse->resourceUri;
 274         }
 275
 276         std::thread exec(context->errorCallback, resourceURI, result);
 277         exec.detach();
 278         return OC_STACK_KEEP_TRANSACTION;
 279     }
 280
 281     OCStackResult InProcClientWrapper::ListenForResource(
 282             const std::string& serviceUrl,
 283             const std::string& resourceType,
 284             OCConnectivityType connectivityType,
 285             FindCallback& callback, QualityOfService QoS)
 286     {
 287         if (!callback)
 288         {
 289             return OC_STACK_INVALID_PARAM;
 290         }
 291
 292         OCStackResult result;
 293         ostringstream resourceUri;
 294         resourceUri << serviceUrl << resourceType;
 295
 296         ClientCallbackContext::ListenContext* context =
 297             new ClientCallbackContext::ListenContext(callback, shared_from_this());
 298         OCCallbackData cbdata;
 299         cbdata.context = static_cast<void*>(context),
 300         cbdata.cb      = listenCallback;
 301         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext*)c;};
 302
 303         auto cLock = m_csdkLock.lock();
 304         if (cLock)
 305         {
 306             std::lock_guard<std::recursive_mutex> lock(*cLock);
 307             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 308                                   resourceUri.str().c_str(),
 309                                   nullptr, nullptr, connectivityType,
 310                                   static_cast<OCQualityOfService>(QoS),
 311                                   &cbdata,
 312                                   nullptr, 0);
 313         }
 314         else
 315         {
 316             delete context;
 317             result = OC_STACK_ERROR;
 318         }
 319         return result;
 320     }
 321
 322     OCStackResult InProcClientWrapper::ListenErrorForResource(
 323             const std::string& serviceUrl,
 324             const std::string& resourceType,
 325             OCConnectivityType connectivityType,
 326             FindCallback& callback, FindErrorCallback& errorCallback,
 327             QualityOfService QoS)
 328     {
 329         if (!callback)
 330         {
 331             return OC_STACK_INVALID_PARAM;
 332         }
 333
 334         ostringstream resourceUri;
 335         resourceUri << serviceUrl << resourceType;
 336
 337         ClientCallbackContext::ListenErrorContext* context =
 338             new ClientCallbackContext::ListenErrorContext(callback, errorCallback,
 339                                                           shared_from_this());
 340         if (!context)
 341         {
 342             return OC_STACK_ERROR;
 343         }
 344
 345         OCCallbackData cbdata(
 346                 static_cast<void*>(context),
 347                 listenErrorCallback,
 348                 [](void* c){delete static_cast<ClientCallbackContext::ListenErrorContext*>(c);}
 349             );
 350
 351         OCStackResult result;
 352         auto cLock = m_csdkLock.lock();
 353         if (cLock)
 354         {
 355             std::lock_guard<std::recursive_mutex> lock(*cLock);
 356             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 357                                   resourceUri.str().c_str(),
 358                                   nullptr, nullptr, connectivityType,
 359                                   static_cast<OCQualityOfService>(QoS),
 360                                   &cbdata,
 361                                   nullptr, 0);
 362         }
 363         else
 364         {
 365             delete context;
 366             result = OC_STACK_ERROR;
 367         }
 368         return result;
 369     }
 370
 371     OCStackApplicationResult listenResListCallback(void* ctx, OCDoHandle /*handle*/,
 372         OCClientResponse* clientResponse)
 373     {
 374         if (!ctx || !clientResponse)
 375         {
 376             return OC_STACK_KEEP_TRANSACTION;
 377         }
 378
 379         ClientCallbackContext::ListenResListContext* context =
 380             static_cast<ClientCallbackContext::ListenResListContext*>(ctx);
 381
 382         if (clientResponse->result != OC_STACK_OK)
 383         {
 384             oclog() << "listenResListCallback(): failed to create resource. clientResponse: "
 385                     << clientResponse->result
 386                     << std::flush;
 387
 388             return OC_STACK_KEEP_TRANSACTION;
 389         }
 390
 391         if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 392         {
 393             oclog() << "listenResListCallback(): clientResponse payload was null or the wrong type"
 394                 << std::flush;
 395             return OC_STACK_KEEP_TRANSACTION;
 396         }
 397
 398         auto clientWrapper = context->clientWrapper.lock();
 399
 400         if (!clientWrapper)
 401         {
 402             oclog() << "listenResListCallback(): failed to get a shared_ptr to the client wrapper"
 403                     << std::flush;
 404             return OC_STACK_KEEP_TRANSACTION;
 405         }
 406
 407         try
 408         {
 409             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 410                                     reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 411
 412             OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 413             std::thread exec(context->callback, container.Resources());
 414             exec.detach();
 415         }
 416         catch (std::exception &e)
 417         {
 418             oclog() << "Exception in listenResListCallback(), ignoring response: "
 419                     << e.what() << std::flush;
 420         }
 421
 422         return OC_STACK_KEEP_TRANSACTION;
 423     }
 424
 425     OCStackResult InProcClientWrapper::ListenForResourceList(
 426             const std::string& serviceUrl,
 427             const std::string& resourceType,
 428             OCConnectivityType connectivityType,
 429             FindResListCallback& callback, QualityOfService QoS)
 430     {
 431         if (!callback)
 432         {
 433             return OC_STACK_INVALID_PARAM;
 434         }
 435
 436         OCStackResult result;
 437         ostringstream resourceUri;
 438         resourceUri << serviceUrl << resourceType;
 439
 440         ClientCallbackContext::ListenResListContext* context =
 441             new ClientCallbackContext::ListenResListContext(callback, shared_from_this());
 442         OCCallbackData cbdata;
 443         cbdata.context = static_cast<void*>(context),
 444         cbdata.cb      = listenResListCallback;
 445         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenResListContext*)c;};
 446
 447         auto cLock = m_csdkLock.lock();
 448         if (cLock)
 449         {
 450             std::lock_guard<std::recursive_mutex> lock(*cLock);
 451             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 452                                   resourceUri.str().c_str(),
 453                                   nullptr, nullptr, connectivityType,
 454                                   static_cast<OCQualityOfService>(QoS),
 455                                   &cbdata,
 456                                   nullptr, 0);
 457         }
 458         else
 459         {
 460             delete context;
 461             result = OC_STACK_ERROR;
 462         }
 463         return result;
 464     }
 465
 466     OCStackApplicationResult listenResListWithErrorCallback(void* ctx, OCDoHandle /*handle*/,
 467         OCClientResponse* clientResponse)
 468     {
 469         if (!ctx || !clientResponse)
 470         {
 471             return OC_STACK_KEEP_TRANSACTION;
 472         }
 473
 474         ClientCallbackContext::ListenResListWithErrorContext* context =
 475                 static_cast<ClientCallbackContext::ListenResListWithErrorContext*>(ctx);
 476
 477         OCStackResult result = clientResponse->result;
 478         if (result != OC_STACK_OK)
 479         {
 480             oclog() << "listenResListWithErrorCallback(): failed to create resource. clientResponse: "
 481                     << result << std::flush;
 482
 483             //send the error callback
 484             std::string uri;
 485             if(NULL != clientResponse->resourceUri)
 486             {
 487                 uri = clientResponse->resourceUri;
 488             }
 489             std::thread exec(context->errorCallback, uri, result);
 490             exec.detach();
 491             return OC_STACK_KEEP_TRANSACTION;
 492         }
 493
 494         if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
 495         {
 496             oclog() << "listenResListWithErrorCallback(): clientResponse payload was null or the wrong type"
 497                 << std::flush;
 498             return OC_STACK_KEEP_TRANSACTION;
 499         }
 500
 501         auto clientWrapper = context->clientWrapper.lock();
 502
 503         if (!clientWrapper)
 504         {
 505             oclog() << "listenResListWithErrorCallback(): failed to get a shared_ptr to the client wrapper"
 506                     << std::flush;
 507             return OC_STACK_KEEP_TRANSACTION;
 508         }
 509
 510         try
 511         {
 512             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 513                             reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
 514
 515             OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 516             std::thread exec(context->callback, container.Resources());
 517             exec.detach();
 518         }
 519         catch (std::exception &e)
 520         {
 521             oclog() << "Exception in listenResListWithErrorCallback(), ignoring response: "
 522             << e.what() << std::flush;
 523         }
 524
 525         return OC_STACK_KEEP_TRANSACTION;
 526     }
 527
 528     OCStackResult InProcClientWrapper::ListenForResourceListWithError(
 529             const std::string& serviceUrl,
 530             const std::string& resourceType,
 531             OCConnectivityType connectivityType,
 532             FindResListCallback& callback,
 533             FindErrorCallback& errorCallback, QualityOfService QoS)
 534     {
 535         if (!callback)
 536         {
 537             return OC_STACK_INVALID_PARAM;
 538         }
 539
 540         OCStackResult result;
 541         ostringstream resourceUri;
 542         resourceUri << serviceUrl << resourceType;
 543
 544         ClientCallbackContext::ListenResListWithErrorContext* context =
 545             new ClientCallbackContext::ListenResListWithErrorContext(callback, errorCallback,
 546                                                           shared_from_this());
 547         if (!context)
 548         {
 549             return OC_STACK_ERROR;
 550         }
 551
 552         OCCallbackData cbdata;
 553         cbdata.context = static_cast<void*>(context),
 554         cbdata.cb      = listenResListWithErrorCallback;
 555         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenResListWithErrorContext*)c;};
 556
 557         auto cLock = m_csdkLock.lock();
 558         if (cLock)
 559         {
 560             std::lock_guard<std::recursive_mutex> lock(*cLock);
 561             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 562                                   resourceUri.str().c_str(),
 563                                   nullptr, nullptr, connectivityType,
 564                                   static_cast<OCQualityOfService>(QoS),
 565                                   &cbdata,
 566                                   nullptr, 0);
 567         }
 568         else
 569         {
 570             delete context;
 571             result = OC_STACK_ERROR;
 572         }
 573         return result;
 574     }
 575
 576 #ifdef WITH_MQ
 577     OCStackApplicationResult listenMQCallback(void* ctx, OCDoHandle /*handle*/,
 578                                               OCClientResponse* clientResponse)
 579     {
 580         ClientCallbackContext::MQTopicContext* context =
 581             static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
 582
 583         if (!clientResponse || !context)
 584         {
 585             return OC_STACK_DELETE_TRANSACTION;
 586         }
 587
 588         std::string resourceURI;
 589         if(NULL != clientResponse->resourceUri)
 590         {
 591             resourceURI  = clientResponse->resourceUri;
 592         }
 593
 594         if (clientResponse->result != OC_STACK_OK)
 595         {
 596             oclog() << "listenMQCallback(): failed to create resource. clientResponse: "
 597                     << clientResponse->result
 598                     << std::flush;
 599
 600             std::thread exec(context->callback, clientResponse->result,
 601                              resourceURI, nullptr);
 602             exec.detach();
 603
 604             return OC_STACK_DELETE_TRANSACTION;
 605         }
 606
 607         auto clientWrapper = context->clientWrapper.lock();
 608         if (!clientWrapper)
 609         {
 610             oclog() << "listenMQCallback(): failed to get a shared_ptr to the client wrapper"
 611                     << std::flush;
 612             return OC_STACK_DELETE_TRANSACTION;
 613         }
 614
 615         try
 616         {
 617             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 618                                         (OCRepPayload *) clientResponse->payload);
 619
 620             // loop to ensure valid construction of all resources
 621             for (auto resource : container.Resources())
 622             {
 623                 std::thread exec(context->callback, clientResponse->result,
 624                                  resourceURI, resource);
 625                 exec.detach();
 626             }
 627         }
 628         catch (std::exception &e)
 629         {
 630             oclog() << "Exception in listCallback, ignoring response: "
 631                     << e.what() << std::flush;
 632         }
 633
 634         return OC_STACK_DELETE_TRANSACTION;
 635     }
 636
 637     OCStackResult InProcClientWrapper::ListenForMQTopic(const OCDevAddr& devAddr,
 638                                                         const std::string& resourceUri,
 639                                                         const QueryParamsMap& queryParams,
 640                                                         const HeaderOptions& headerOptions,
 641                                                         MQTopicCallback& callback,
 642                                                         QualityOfService QoS)
 643     {
 644         oclog() << "ListenForMQTopic()" << std::flush;
 645
 646         if (!callback)
 647         {
 648             return OC_STACK_INVALID_PARAM;
 649         }
 650
 651         if (headerOptions.size() > MAX_HEADER_OPTIONS)
 652         {
 653             oclog() << "ListenForMQTopic: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
 654             return OC_STACK_INVALID_PARAM;
 655         }
 656
 657         ClientCallbackContext::MQTopicContext* context =
 658             new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 659         OCCallbackData cbdata;
 660         cbdata.context = static_cast<void*>(context),
 661         cbdata.cb      = listenMQCallback;
 662         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 663
 664         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 665
 666         OCStackResult result = OC_STACK_ERROR;
 667         auto cLock = m_csdkLock.lock();
 668         if (cLock)
 669         {
 670             std::lock_guard<std::recursive_mutex> lock(*cLock);
 671             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
 672             result = OCDoResource(
 673                                   nullptr, OC_REST_GET,
 674                                   uri.c_str(),
 675                                   &devAddr, nullptr,
 676                                   CT_DEFAULT,
 677                                   static_cast<OCQualityOfService>(QoS),
 678                                   &cbdata,
 679                                   options,
 680                                   headerOptions.size());
 681             delete[] options;
 682         }
 683         else
 684         {
 685             delete context;
 686         }
 687
 688         return result;
 689     }
 690 #endif
 691
 692     OCStackApplicationResult listenDeviceCallback(void* ctx,
 693                                                   OCDoHandle /*handle*/,
 694             OCClientResponse* clientResponse)
 695     {
 696         ClientCallbackContext::DeviceListenContext* context =
 697             static_cast<ClientCallbackContext::DeviceListenContext*>(ctx);
 698
 699         try
 700         {
 701             OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 702             OCRepresentation rep = parseGetSetCallback(clientResponse);
 703             std::thread exec(context->callback, rep);
 704             exec.detach();
 705         }
 706         catch(OC::OCException& e)
 707         {
 708             oclog() <<"Exception in listenDeviceCallback, ignoring response: "
 709                 <<e.what() <<std::flush;
 710         }
 711
 712         return OC_STACK_KEEP_TRANSACTION;
 713     }
 714
 715     OCStackResult InProcClientWrapper::ListenForDevice(
 716             const std::string& serviceUrl,
 717             const std::string& deviceURI,
 718             OCConnectivityType connectivityType,
 719             FindDeviceCallback& callback,
 720             QualityOfService QoS)
 721     {
 722         if (!callback)
 723         {
 724             return OC_STACK_INVALID_PARAM;
 725         }
 726         OCStackResult result;
 727         ostringstream deviceUri;
 728         deviceUri << serviceUrl << deviceURI;
 729
 730         ClientCallbackContext::DeviceListenContext* context =
 731             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
 732         OCCallbackData cbdata;
 733
 734         cbdata.context = static_cast<void*>(context),
 735         cbdata.cb      = listenDeviceCallback;
 736         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
 737
 738         auto cLock = m_csdkLock.lock();
 739         if (cLock)
 740         {
 741             std::lock_guard<std::recursive_mutex> lock(*cLock);
 742             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 743                                   deviceUri.str().c_str(),
 744                                   nullptr, nullptr, connectivityType,
 745                                   static_cast<OCQualityOfService>(QoS),
 746                                   &cbdata,
 747                                   nullptr, 0);
 748         }
 749         else
 750         {
 751             delete context;
 752             result = OC_STACK_ERROR;
 753         }
 754         return result;
 755     }
 756
 757     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 758                     HeaderOptions& serverHeaderOptions)
 759     {
 760         if (clientResponse)
 761         {
 762             // Parse header options from server
 763             uint16_t optionID;
 764             std::string optionData;
 765
 766             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 767             {
 768                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 769                 optionData = reinterpret_cast<const char*>
 770                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 771                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 772                 serverHeaderOptions.push_back(headerOption);
 773             }
 774         }
 775         else
 776         {
 777             // clientResponse is invalid
 778             // TODO check proper logging
 779             std::cout << " Invalid response " << std::endl;
 780         }
 781     }
 782
 783 #ifdef WITH_MQ
 784     OCStackApplicationResult createMQTopicCallback(void* ctx, OCDoHandle /*handle*/,
 785                     OCClientResponse* clientResponse)
 786     {
 787         ClientCallbackContext::MQTopicContext* context =
 788             static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
 789         HeaderOptions serverHeaderOptions;
 790
 791         if (!clientResponse || !context)
 792         {
 793             return OC_STACK_DELETE_TRANSACTION;
 794         }
 795
 796         std::string createdUri;
 797         bool isLocationOption = false;
 798         OCStackResult result = clientResponse->result;
 799         if (OC_STACK_OK               == result ||
 800             OC_STACK_RESOURCE_CREATED == result)
 801         {
 802             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 803
 804             for (auto headerOption : serverHeaderOptions)
 805             {
 806                 if (HeaderOption::LOCATION_PATH_OPTION_ID == headerOption.getOptionID())
 807                 {
 808                     createdUri += "/";
 809                     createdUri += headerOption.getOptionData();
 810                     if (!isLocationOption)
 811                     {
 812                         isLocationOption = true;
 813                     }
 814                 }
 815             }
 816         }
 817
 818         if (!isLocationOption && NULL != clientResponse->resourceUri)
 819         {
 820             createdUri = std::string(clientResponse->resourceUri);
 821         }
 822
 823         auto clientWrapper = context->clientWrapper.lock();
 824
 825         if (!clientWrapper)
 826         {
 827             oclog() << "createMQTopicCallback(): failed to get a shared_ptr to the client wrapper"
 828                     << std::flush;
 829             return OC_STACK_DELETE_TRANSACTION;
 830         }
 831
 832         try
 833         {
 834             if (OC_STACK_OK               == result ||
 835                 OC_STACK_RESOURCE_CREATED == result)
 836             {
 837                 ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 838                                             createdUri);
 839                 for (auto resource : container.Resources())
 840                 {
 841                     std::thread exec(context->callback, result,
 842                                      createdUri,
 843                                      resource);
 844                     exec.detach();
 845                 }
 846             }
 847             else
 848             {
 849                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 850                 std::thread exec(context->callback, result,
 851                                  createdUri,
 852                                  nullptr);
 853                 exec.detach();
 854             }
 855         }
 856         catch (std::exception &e)
 857         {
 858             oclog() << "Exception in createMQTopicCallback, ignoring response: "
 859                     << e.what() << std::flush;
 860         }
 861         return OC_STACK_DELETE_TRANSACTION;
 862     }
 863
 864     OCStackResult InProcClientWrapper::PutMQTopicRepresentation(
 865                 const OCDevAddr& devAddr,
 866                 const std::string& uri,
 867                 const OCRepresentation& rep,
 868                 const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 869                 MQTopicCallback& callback, QualityOfService QoS)
 870     {
 871         if (!callback)
 872         {
 873             return OC_STACK_INVALID_PARAM;
 874         }
 875
 876         if (headerOptions.size() > MAX_HEADER_OPTIONS)
 877         {
 878             oclog() << "PutMQTopicRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
 879             return OC_STACK_INVALID_PARAM;
 880         }
 881
 882         OCStackResult result;
 883         ClientCallbackContext::MQTopicContext* ctx =
 884                 new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 885         OCCallbackData cbdata;
 886         cbdata.context = static_cast<void*>(ctx),
 887         cbdata.cb      = createMQTopicCallback;
 888         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 889
 890         std::string url = assembleSetResourceUri(uri, queryParams);
 891
 892         auto cLock = m_csdkLock.lock();
 893
 894         if (cLock)
 895         {
 896             std::lock_guard<std::recursive_mutex> lock(*cLock);
 897             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
 898
 899             result = OCDoResource(nullptr, OC_REST_PUT,
 900                                   url.c_str(), &devAddr,
 901                                   assembleSetResourcePayload(rep),
 902                                   CT_DEFAULT,
 903                                   static_cast<OCQualityOfService>(QoS),
 904                                   &cbdata,
 905                                   options,
 906                                   headerOptions.size());
 907              delete[] options;
 908         }
 909         else
 910         {
 911             delete ctx;
 912             result = OC_STACK_ERROR;
 913         }
 914
 915         return result;
 916     }
 917 #endif
 918     OCStackApplicationResult getResourceCallback(void* ctx,
 919                                                  OCDoHandle /*handle*/,
 920         OCClientResponse* clientResponse)
 921     {
 922         ClientCallbackContext::GetContext* context =
 923             static_cast<ClientCallbackContext::GetContext*>(ctx);
 924         OCRepresentation rep;
 925         HeaderOptions serverHeaderOptions;
 926         OCStackResult result = clientResponse->result;
 927
 928         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 929         try
 930         {
 931             rep = parseGetSetCallback(clientResponse);
 932         }
 933         catch(OC::OCException& e)
 934         {
 935             result = e.code();
 936         }
 937
 938         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 939         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 940         exec.detach();
 941         return OC_STACK_DELETE_TRANSACTION;
 942     }
 943
 944     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 945         const OCDevAddr& devAddr,
 946         const std::string& resourceUri,
 947         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 948         OCConnectivityType connectivityType,
 949         GetCallback& callback, QualityOfService QoS)
 950     {
 951         if (!callback)
 952         {
 953             return OC_STACK_INVALID_PARAM;
 954         }
 955
 956         if (headerOptions.size() > MAX_HEADER_OPTIONS)
 957         {
 958             oclog() << "GetResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
 959             return OC_STACK_INVALID_PARAM;
 960         }
 961
 962         OCStackResult result;
 963         ClientCallbackContext::GetContext* ctx =
 964             new ClientCallbackContext::GetContext(callback);
 965
 966         OCCallbackData cbdata;
 967         cbdata.context = static_cast<void*>(ctx);
 968         cbdata.cb      = getResourceCallback;
 969         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 970
 971         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 972
 973         auto cLock = m_csdkLock.lock();
 974
 975         if (cLock)
 976         {
 977             std::lock_guard<std::recursive_mutex> lock(*cLock);
 978             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
 979
 980             result = OCDoResource(
 981                                   nullptr, OC_REST_GET,
 982                                   uri.c_str(),
 983                                   &devAddr, nullptr,
 984                                   connectivityType,
 985                                   static_cast<OCQualityOfService>(QoS),
 986                                   &cbdata,
 987                                   options,
 988                                   headerOptions.size());
 989             delete[] options;
 990         }
 991         else
 992         {
 993             delete ctx;
 994             result = OC_STACK_ERROR;
 995         }
 996         return result;
 997     }
 998
 999
1000     OCStackApplicationResult setResourceCallback(void* ctx,
1001                                                  OCDoHandle /*handle*/,
1002         OCClientResponse* clientResponse)
1003     {
1004         ClientCallbackContext::SetContext* context =
1005             static_cast<ClientCallbackContext::SetContext*>(ctx);
1006         OCRepresentation attrs;
1007         HeaderOptions serverHeaderOptions;
1008
1009         OCStackResult result = clientResponse->result;
1010
1011         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1012         try
1013         {
1014             attrs = parseGetSetCallback(clientResponse);
1015         }
1016         catch(OC::OCException& e)
1017         {
1018             result = e.code();
1019         }
1020
1021         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1022         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
1023         exec.detach();
1024         return OC_STACK_DELETE_TRANSACTION;
1025     }
1026
1027     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1028         const QueryParamsMap& queryParams)
1029     {
1030         if (!uri.empty())
1031         {
1032             if (uri.back() == '/')
1033             {
1034                 uri.resize(uri.size() - 1);
1035             }
1036         }
1037
1038         ostringstream paramsList;
1039         if (queryParams.size() > 0)
1040         {
1041             paramsList << '?';
1042         }
1043
1044         for (auto& param : queryParams)
1045         {
1046             paramsList << param.first <<'='<<param.second<<';';
1047         }
1048
1049         std::string queryString = paramsList.str();
1050
1051         if (queryString.empty())
1052         {
1053             return uri;
1054         }
1055
1056         if (queryString.back() == ';')
1057         {
1058             queryString.resize(queryString.size() - 1);
1059         }
1060
1061         std::string ret = uri + queryString;
1062         return ret;
1063     }
1064
1065     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1066         const QueryParamsList& queryParams)
1067     {
1068         if (!uri.empty())
1069         {
1070             if (uri.back() == '/')
1071             {
1072                 uri.resize(uri.size() - 1);
1073             }
1074         }
1075
1076         ostringstream paramsList;
1077         if (queryParams.size() > 0)
1078         {
1079             paramsList << '?';
1080         }
1081
1082         for (auto& param : queryParams)
1083         {
1084             for (auto& paramList : param.second)
1085             {
1086                 paramsList << param.first << '=' << paramList << ';';
1087             }
1088         }
1089
1090         std::string queryString = paramsList.str();
1091
1092         if (queryString.empty())
1093         {
1094             return uri;
1095         }
1096
1097         if (queryString.back() == ';')
1098         {
1099             queryString.resize(queryString.size() - 1);
1100         }
1101
1102         std::string ret = uri + queryString;
1103         return ret;
1104     }
1105
1106     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
1107     {
1108         MessageContainer ocInfo;
1109         ocInfo.addRepresentation(rep);
1110         for(const OCRepresentation& r : rep.getChildren())
1111         {
1112             ocInfo.addRepresentation(r);
1113         }
1114
1115         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
1116     }
1117
1118     OCStackResult InProcClientWrapper::PostResourceRepresentation(
1119         const OCDevAddr& devAddr,
1120         const std::string& uri,
1121         const OCRepresentation& rep,
1122         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1123         OCConnectivityType connectivityType,
1124         PostCallback& callback, QualityOfService QoS)
1125     {
1126         if (!callback)
1127         {
1128             return OC_STACK_INVALID_PARAM;
1129         }
1130
1131         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1132         {
1133             oclog() << "PostResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1134             return OC_STACK_INVALID_PARAM;
1135         }
1136
1137         OCStackResult result;
1138         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1139         OCCallbackData cbdata;
1140         cbdata.context = static_cast<void*>(ctx),
1141         cbdata.cb      = setResourceCallback;
1142         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1143
1144
1145         std::string url = assembleSetResourceUri(uri, queryParams);
1146
1147         auto cLock = m_csdkLock.lock();
1148
1149         if (cLock)
1150         {
1151             std::lock_guard<std::recursive_mutex> lock(*cLock);
1152             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1153
1154             result = OCDoResource(nullptr, OC_REST_POST,
1155                                   url.c_str(), &devAddr,
1156                                   assembleSetResourcePayload(rep),
1157                                   connectivityType,
1158                                   static_cast<OCQualityOfService>(QoS),
1159                                   &cbdata,
1160                                   options,
1161                                   headerOptions.size());
1162             delete[] options;
1163         }
1164         else
1165         {
1166             delete ctx;
1167             result = OC_STACK_ERROR;
1168         }
1169
1170         return result;
1171     }
1172
1173     OCStackResult InProcClientWrapper::PutResourceRepresentation(
1174         const OCDevAddr& devAddr,
1175         const std::string& uri,
1176         const OCRepresentation& rep,
1177         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1178         PutCallback& callback, QualityOfService QoS)
1179     {
1180         if (!callback)
1181         {
1182             return OC_STACK_INVALID_PARAM;
1183         }
1184
1185         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1186         {
1187             oclog() << "PutResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1188             return OC_STACK_INVALID_PARAM;
1189         }
1190
1191         OCStackResult result;
1192         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1193         OCCallbackData cbdata;
1194         cbdata.context = static_cast<void*>(ctx),
1195         cbdata.cb      = setResourceCallback;
1196         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1197
1198
1199         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1200
1201         auto cLock = m_csdkLock.lock();
1202
1203         if (cLock)
1204         {
1205             std::lock_guard<std::recursive_mutex> lock(*cLock);
1206             OCDoHandle handle;
1207             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1208
1209             result = OCDoResource(&handle, OC_REST_PUT,
1210                                   url.c_str(), &devAddr,
1211                                   assembleSetResourcePayload(rep),
1212                                   CT_DEFAULT,
1213                                   static_cast<OCQualityOfService>(QoS),
1214                                   &cbdata,
1215                                   options,
1216                                   headerOptions.size());
1217             delete[] options;
1218         }
1219         else
1220         {
1221             delete ctx;
1222             result = OC_STACK_ERROR;
1223         }
1224
1225         return result;
1226     }
1227
1228     OCStackApplicationResult deleteResourceCallback(void* ctx,
1229                                                     OCDoHandle /*handle*/,
1230         OCClientResponse* clientResponse)
1231     {
1232         ClientCallbackContext::DeleteContext* context =
1233             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
1234         HeaderOptions serverHeaderOptions;
1235
1236         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1237
1238         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1239         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
1240         exec.detach();
1241         return OC_STACK_DELETE_TRANSACTION;
1242     }
1243
1244     OCStackResult InProcClientWrapper::DeleteResource(
1245         const OCDevAddr& devAddr,
1246         const std::string& uri,
1247         const HeaderOptions& headerOptions,
1248         OCConnectivityType connectivityType,
1249         DeleteCallback& callback,
1250         QualityOfService /*QoS*/)
1251     {
1252         if (!callback)
1253         {
1254             return OC_STACK_INVALID_PARAM;
1255         }
1256
1257         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1258         {
1259             oclog() << "DeleteResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1260             return OC_STACK_INVALID_PARAM;
1261         }
1262
1263         OCStackResult result;
1264         ClientCallbackContext::DeleteContext* ctx =
1265             new ClientCallbackContext::DeleteContext(callback);
1266         OCCallbackData cbdata;
1267         cbdata.context = static_cast<void*>(ctx),
1268         cbdata.cb      = deleteResourceCallback;
1269         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
1270
1271
1272         auto cLock = m_csdkLock.lock();
1273
1274         if (cLock)
1275         {
1276             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1277
1278             std::lock_guard<std::recursive_mutex> lock(*cLock);
1279
1280             result = OCDoResource(nullptr, OC_REST_DELETE,
1281                                   uri.c_str(), &devAddr,
1282                                   nullptr,
1283                                   connectivityType,
1284                                   static_cast<OCQualityOfService>(m_cfg.QoS),
1285                                   &cbdata,
1286                                   options,
1287                                   headerOptions.size());
1288             delete[] options;
1289         }
1290         else
1291         {
1292             delete ctx;
1293             result = OC_STACK_ERROR;
1294         }
1295
1296         return result;
1297     }
1298
1299     OCStackApplicationResult observeResourceCallback(void* ctx,
1300                                                      OCDoHandle /*handle*/,
1301         OCClientResponse* clientResponse)
1302     {
1303         ClientCallbackContext::ObserveContext* context =
1304             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
1305         OCRepresentation attrs;
1306         HeaderOptions serverHeaderOptions;
1307         uint32_t sequenceNumber = clientResponse->sequenceNumber;
1308         OCStackResult result = clientResponse->result;
1309
1310         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1311         try
1312         {
1313             attrs = parseGetSetCallback(clientResponse);
1314         }
1315         catch(OC::OCException& e)
1316         {
1317             result = e.code();
1318         }
1319
1320         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1321         std::thread exec(context->callback, serverHeaderOptions, attrs,
1322                     result, sequenceNumber);
1323         exec.detach();
1324         if (sequenceNumber == MAX_SEQUENCE_NUMBER + 1)
1325         {
1326             return OC_STACK_DELETE_TRANSACTION;
1327         }
1328
1329         return OC_STACK_KEEP_TRANSACTION;
1330     }
1331
1332     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
1333         const OCDevAddr& devAddr,
1334         const std::string& uri,
1335         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1336         ObserveCallback& callback, QualityOfService QoS)
1337     {
1338         if (!callback)
1339         {
1340             return OC_STACK_INVALID_PARAM;
1341         }
1342
1343         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1344         {
1345             oclog() << "ObserveResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1346             return OC_STACK_INVALID_PARAM;
1347         }
1348
1349         OCStackResult result;
1350
1351         ClientCallbackContext::ObserveContext* ctx =
1352             new ClientCallbackContext::ObserveContext(callback);
1353         OCCallbackData cbdata;
1354         cbdata.context = static_cast<void*>(ctx),
1355         cbdata.cb      = observeResourceCallback;
1356         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1357
1358
1359         OCMethod method;
1360         if (observeType == ObserveType::Observe)
1361         {
1362             method = OC_REST_OBSERVE;
1363         }
1364         else if (observeType == ObserveType::ObserveAll)
1365         {
1366             method = OC_REST_OBSERVE_ALL;
1367         }
1368         else
1369         {
1370             method = OC_REST_OBSERVE_ALL;
1371         }
1372
1373         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1374
1375         auto cLock = m_csdkLock.lock();
1376
1377         if (cLock)
1378         {
1379             std::lock_guard<std::recursive_mutex> lock(*cLock);
1380             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1381
1382             result = OCDoResource(handle, method,
1383                                   url.c_str(), &devAddr,
1384                                   nullptr,
1385                                   CT_DEFAULT,
1386                                   static_cast<OCQualityOfService>(QoS),
1387                                   &cbdata,
1388                                   options,
1389                                   headerOptions.size());
1390             delete[] options;
1391         }
1392         else
1393         {
1394             delete ctx;
1395             return OC_STACK_ERROR;
1396         }
1397
1398         return result;
1399     }
1400
1401     OCStackResult InProcClientWrapper::CancelObserveResource(
1402             OCDoHandle handle,
1403             const std::string& /*host*/,
1404             const std::string& /*uri*/,
1405             const HeaderOptions& headerOptions,
1406             QualityOfService QoS)
1407     {
1408         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1409         {
1410             oclog() << "CancelObserveResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1411             return OC_STACK_INVALID_PARAM;
1412         }
1413
1414         OCStackResult result;
1415         auto cLock = m_csdkLock.lock();
1416
1417         if (cLock)
1418         {
1419             std::lock_guard<std::recursive_mutex> lock(*cLock);
1420             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1421
1422             result = OCCancel(handle,
1423                     static_cast<OCQualityOfService>(QoS),
1424                     options,
1425                     headerOptions.size());
1426             delete[] options;
1427         }
1428         else
1429         {
1430             result = OC_STACK_ERROR;
1431         }
1432
1433         return result;
1434     }
1435
1436 #ifdef WITH_PRESENCE
1437     OCStackApplicationResult subscribePresenceCallback(void* ctx,
1438                                                        OCDoHandle /*handle*/,
1439             OCClientResponse* clientResponse)
1440     {
1441         ClientCallbackContext::SubscribePresenceContext* context =
1442         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
1443
1444         /*
1445          * This a hack while we rethink presence subscription.
1446          */
1447         std::string url = clientResponse->devAddr.addr;
1448
1449         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1450         std::thread exec(context->callback, clientResponse->result,
1451                     clientResponse->sequenceNumber, url);
1452
1453         exec.detach();
1454
1455         return OC_STACK_KEEP_TRANSACTION;
1456     }
1457 #endif
1458
1459     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
1460         const std::string& host, const std::string& resourceType,
1461         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
1462     {
1463 #ifdef WITH_PRESENCE
1464         if (!presenceHandler)
1465         {
1466             return OC_STACK_INVALID_PARAM;
1467         }
1468
1469         ClientCallbackContext::SubscribePresenceContext* ctx =
1470             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
1471         OCCallbackData cbdata;
1472         cbdata.context = static_cast<void*>(ctx),
1473         cbdata.cb      = subscribePresenceCallback;
1474         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
1475
1476
1477         auto cLock = m_csdkLock.lock();
1478
1479         std::ostringstream os;
1480         os << host << OC_RSRVD_PRESENCE_URI;
1481
1482         if (!resourceType.empty())
1483         {
1484             os << "?rt=" << resourceType;
1485         }
1486
1487         if (!cLock)
1488         {
1489             delete ctx;
1490             return OC_STACK_ERROR;
1491         }
1492
1493         return OCDoResource(handle, OC_REST_PRESENCE,
1494                             os.str().c_str(), nullptr,
1495                             nullptr, connectivityType,
1496                             OC_LOW_QOS, &cbdata, NULL, 0);
1497 #else
1498         return OC_STACK_NOT_IMPLEMENTED;
1499 #endif
1500     }
1501
1502     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
1503     {
1504 #ifdef WITH_PRESENCE
1505         OCStackResult result;
1506         auto cLock = m_csdkLock.lock();
1507
1508         if (cLock)
1509         {
1510             std::lock_guard<std::recursive_mutex> lock(*cLock);
1511             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
1512         }
1513         else
1514         {
1515             result = OC_STACK_ERROR;
1516         }
1517
1518         return result;
1519 #else
1520         return OC_STACK_NOT_IMPLEMENTED;
1521 #endif
1522     }
1523
1524 #ifdef WITH_CLOUD
1525     OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
1526                                                                const std::string& host,
1527                                                                const std::vector<std::string>& di,
1528                                                                OCConnectivityType connectivityType,
1529                                                                ObserveCallback& callback)
1530     {
1531         if (!callback)
1532         {
1533             return OC_STACK_INVALID_PARAM;
1534         }
1535         OCStackResult result;
1536
1537         ClientCallbackContext::ObserveContext* ctx =
1538             new ClientCallbackContext::ObserveContext(callback);
1539         OCCallbackData cbdata;
1540         cbdata.context = static_cast<void*>(ctx),
1541         cbdata.cb      = observeResourceCallback;
1542         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1543
1544         auto cLock = m_csdkLock.lock();
1545
1546         if (cLock)
1547         {
1548             std::lock_guard<std::recursive_mutex> lock(*cLock);
1549
1550             std::ostringstream os;
1551             os << host << OC_RSRVD_DEVICE_PRESENCE_URI;
1552             QueryParamsList queryParams({{OC_RSRVD_DEVICE_ID, di}});
1553             std::string url = assembleSetResourceUri(os.str(), queryParams);
1554
1555             result = OCDoResource(handle, OC_REST_OBSERVE,
1556                                   url.c_str(), nullptr,
1557                                   nullptr, connectivityType,
1558                                   OC_LOW_QOS, &cbdata,
1559                                   nullptr, 0);
1560         }
1561         else
1562         {
1563             delete ctx;
1564             result = OC_STACK_ERROR;
1565         }
1566
1567         return result;
1568     }
1569 #endif
1570
1571     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
1572     {
1573         qos = m_cfg.QoS;
1574         return OC_STACK_OK;
1575     }
1576
1577     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(const HeaderOptions& headerOptions)
1578     {
1579         if ( headerOptions.size() == 0)
1580         {
1581             return nullptr;
1582         }
1583
1584         OCHeaderOption* options = new OCHeaderOption[headerOptions.size()]();
1585
1586         size_t numOptions = 0;
1587         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
1588         {
1589             OCStackResult ret = OCSetHeaderOption(options, &numOptions, it->getOptionID(),
1590                                     it->getOptionData().c_str(), it->getOptionData().length());
1591             if (OC_STACK_OK != ret)
1592             {
1593                 OIC_LOG_V(ERROR, TAG, "Failed to convert vnd header options! (error=%d)", ret);
1594                 delete[] options;
1595                 return nullptr;
1596             }
1597         }
1598
1599         return options;
1600     }
1601
1602     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
1603     {
1604         if (!dev)
1605         {
1606             return nullptr;
1607         }
1608
1609         OCDPDev_t* result = new OCDPDev_t(*dev);
1610         result->prm = new OCPrm_t[dev->prmLen];
1611         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
1612         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
1613     }
1614
1615     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
1616     {
1617         while(list)
1618         {
1619             dpList.push_back(cloneDevice(list));
1620             list = list->next;
1621         }
1622     }
1623
1624     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
1625             GetDirectPairedCallback& callback)
1626     {
1627         if (!callback || 0 == waittime)
1628         {
1629             return OC_STACK_INVALID_PARAM;
1630         }
1631
1632         OCStackResult result = OC_STACK_ERROR;
1633         const OCDPDev_t *list = nullptr;
1634         PairedDevices dpDeviceList;
1635
1636         auto cLock = m_csdkLock.lock();
1637
1638         if (cLock)
1639         {
1640             std::lock_guard<std::recursive_mutex> lock(*cLock);
1641
1642             list = OCDiscoverDirectPairingDevices(waittime);
1643             if (NULL == list)
1644             {
1645                 result = OC_STACK_NO_RESOURCE;
1646                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1647                     << std::flush;
1648             }
1649             else {
1650                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1651                 convert(list, dpDeviceList);
1652                 std::thread exec(callback, dpDeviceList);
1653                 exec.detach();
1654                 result = OC_STACK_OK;
1655             }
1656         }
1657         else
1658         {
1659             result = OC_STACK_ERROR;
1660         }
1661
1662         return result;
1663     }
1664
1665     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1666     {
1667         if (!callback)
1668         {
1669             return OC_STACK_INVALID_PARAM;
1670         }
1671
1672         OCStackResult result = OC_STACK_ERROR;
1673         const OCDPDev_t *list = nullptr;
1674         PairedDevices dpDeviceList;
1675
1676         auto cLock = m_csdkLock.lock();
1677
1678         if (cLock)
1679         {
1680             std::lock_guard<std::recursive_mutex> lock(*cLock);
1681
1682             list = OCGetDirectPairedDevices();
1683             if (NULL == list)
1684             {
1685                 result = OC_STACK_NO_RESOURCE;
1686                 OIC_LOG_V(DEBUG, TAG, "%s: No device found for direct pairing", __func__);
1687             }
1688             else {
1689                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1690                 convert(list, dpDeviceList);
1691                 std::thread exec(callback, dpDeviceList);
1692                 exec.detach();
1693                 result = OC_STACK_OK;
1694             }
1695         }
1696         else
1697         {
1698             result = OC_STACK_ERROR;
1699         }
1700
1701         return result;
1702     }
1703
1704     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1705             OCStackResult result)
1706     {
1707
1708         ClientCallbackContext::DirectPairingContext* context =
1709             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1710
1711         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1712         std::thread exec(context->callback, cloneDevice(peer), result);
1713         exec.detach();
1714     }
1715
1716     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1717             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1718     {
1719         if (!peer || !callback)
1720         {
1721             oclog() << "Invalid parameters" << std::flush;
1722             return OC_STACK_INVALID_PARAM;
1723         }
1724
1725         OCStackResult result = OC_STACK_ERROR;
1726         ClientCallbackContext::DirectPairingContext* context =
1727             new ClientCallbackContext::DirectPairingContext(callback);
1728
1729         auto cLock = m_csdkLock.lock();
1730         if (cLock)
1731         {
1732             std::lock_guard<std::recursive_mutex> lock(*cLock);
1733             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1734                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1735         }
1736         else
1737         {
1738             delete context;
1739             result = OC_STACK_ERROR;
1740         }
1741         return result;
1742     }
1743 #ifdef TCP_ADAPTER
1744     OCStackApplicationResult KeepAliveRespCallback(void* ctx,
1745                                                  OCDoHandle /*handle*/,
1746         OCClientResponse* clientResponse)
1747     {
1748         ClientCallbackContext::KeepAliveContext* context =
1749             static_cast<ClientCallbackContext::KeepAliveContext*>(ctx);
1750         OCRepresentation attrs;
1751         OCStackResult result = clientResponse->result;
1752
1753         try
1754         {
1755             attrs = parseGetSetCallback(clientResponse);
1756         }
1757         catch(OC::OCException& e)
1758         {
1759             result = e.code();
1760         }
1761
1762         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1763         std::thread exec(context->callback, result, attrs);
1764         exec.detach();
1765         return OC_STACK_DELETE_TRANSACTION;
1766     }
1767
1768     OCStackResult InProcClientWrapper::findKeepAliveResource(std::string host,
1769                                                              KeepAliveCallback resultCallback)
1770     {
1771         if (host.empty() || !resultCallback)
1772         {
1773             oclog() << "Invalid parameters" << std::flush;
1774             return OC_STACK_INVALID_PARAM;
1775         }
1776
1777         OCStackResult result = OC_STACK_ERROR;
1778
1779         ClientCallbackContext::KeepAliveContext* ctx =
1780                        new ClientCallbackContext::KeepAliveContext(resultCallback);
1781         OCCallbackData cbdata;
1782         cbdata.context = static_cast<void*>(ctx),
1783         cbdata.cb      = KeepAliveRespCallback;
1784         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1785
1786         auto cLock = m_csdkLock.lock();
1787
1788         if (cLock)
1789         {
1790             std::lock_guard<std::recursive_mutex> lock(*cLock);
1791             result = OCFindKeepAliveResource(nullptr, host.c_str(), &cbdata);
1792         }
1793         else
1794         {
1795             delete ctx;
1796             result = OC_STACK_ERROR;
1797         }
1798         return result;
1799     }
1800
1801     OCStackResult InProcClientWrapper::sendKeepAliveRequest(std::string host,
1802                                                             const OCRepresentation& rep,
1803                                                             KeepAliveCallback resultCallback)
1804     {
1805         if (!resultCallback)
1806         {
1807             oclog() << "Invalid parameters" << std::flush;
1808             return OC_STACK_INVALID_PARAM;
1809         }
1810
1811         OCStackResult result = OC_STACK_ERROR;
1812
1813         ClientCallbackContext::KeepAliveContext* ctx = new ClientCallbackContext::KeepAliveContext(resultCallback);
1814         OCCallbackData cbdata;
1815         cbdata.context = static_cast<void*>(ctx),
1816         cbdata.cb      = KeepAliveRespCallback;
1817         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1818
1819         auto cLock = m_csdkLock.lock();
1820
1821         if (cLock)
1822         {
1823             std::lock_guard<std::recursive_mutex> lock(*cLock);
1824             OCRepPayload *payload = rep.getPayload();
1825             result = OCSendKeepAliveRequest (nullptr, host.c_str(), (OCPayload*)payload, &cbdata);
1826         }
1827         else
1828         {
1829             delete ctx;
1830             result = OC_STACK_ERROR;
1831         }
1832         return result;
1833     }
1834 #endif
1835 }