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         ClientCallbackContext::MQTopicContext* context =
 652             new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 653         OCCallbackData cbdata;
 654         cbdata.context = static_cast<void*>(context),
 655         cbdata.cb      = listenMQCallback;
 656         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 657
 658         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 659
 660         OCStackResult result = OC_STACK_ERROR;
 661         auto cLock = m_csdkLock.lock();
 662         if (cLock)
 663         {
 664             std::lock_guard<std::recursive_mutex> lock(*cLock);
 665             OCHeaderOption options[MAX_HEADER_OPTIONS];
 666             result = OCDoResource(
 667                                   nullptr, OC_REST_GET,
 668                                   uri.c_str(),
 669                                   &devAddr, nullptr,
 670                                   CT_DEFAULT,
 671                                   static_cast<OCQualityOfService>(QoS),
 672                                   &cbdata,
 673                                   assembleHeaderOptions(options, headerOptions),
 674                                   headerOptions.size());
 675         }
 676         else
 677         {
 678             delete context;
 679         }
 680
 681         return result;
 682     }
 683 #endif
 684
 685     OCStackApplicationResult listenDeviceCallback(void* ctx,
 686                                                   OCDoHandle /*handle*/,
 687             OCClientResponse* clientResponse)
 688     {
 689         ClientCallbackContext::DeviceListenContext* context =
 690             static_cast<ClientCallbackContext::DeviceListenContext*>(ctx);
 691
 692         try
 693         {
 694             OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 695             OCRepresentation rep = parseGetSetCallback(clientResponse);
 696             std::thread exec(context->callback, rep);
 697             exec.detach();
 698         }
 699         catch(OC::OCException& e)
 700         {
 701             oclog() <<"Exception in listenDeviceCallback, ignoring response: "
 702                 <<e.what() <<std::flush;
 703         }
 704
 705         return OC_STACK_KEEP_TRANSACTION;
 706     }
 707
 708     OCStackResult InProcClientWrapper::ListenForDevice(
 709             const std::string& serviceUrl,
 710             const std::string& deviceURI,
 711             OCConnectivityType connectivityType,
 712             FindDeviceCallback& callback,
 713             QualityOfService QoS)
 714     {
 715         if (!callback)
 716         {
 717             return OC_STACK_INVALID_PARAM;
 718         }
 719         OCStackResult result;
 720         ostringstream deviceUri;
 721         deviceUri << serviceUrl << deviceURI;
 722
 723         ClientCallbackContext::DeviceListenContext* context =
 724             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
 725         OCCallbackData cbdata;
 726
 727         cbdata.context = static_cast<void*>(context),
 728         cbdata.cb      = listenDeviceCallback;
 729         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
 730
 731         auto cLock = m_csdkLock.lock();
 732         if (cLock)
 733         {
 734             std::lock_guard<std::recursive_mutex> lock(*cLock);
 735             result = OCDoResource(nullptr, OC_REST_DISCOVER,
 736                                   deviceUri.str().c_str(),
 737                                   nullptr, nullptr, connectivityType,
 738                                   static_cast<OCQualityOfService>(QoS),
 739                                   &cbdata,
 740                                   nullptr, 0);
 741         }
 742         else
 743         {
 744             delete context;
 745             result = OC_STACK_ERROR;
 746         }
 747         return result;
 748     }
 749
 750     void parseServerHeaderOptions(OCClientResponse* clientResponse,
 751                     HeaderOptions& serverHeaderOptions)
 752     {
 753         if (clientResponse)
 754         {
 755             // Parse header options from server
 756             uint16_t optionID;
 757             std::string optionData;
 758
 759             for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
 760             {
 761                 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
 762                 optionData = reinterpret_cast<const char*>
 763                                 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
 764                 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
 765                 serverHeaderOptions.push_back(headerOption);
 766             }
 767         }
 768         else
 769         {
 770             // clientResponse is invalid
 771             // TODO check proper logging
 772             std::cout << " Invalid response " << std::endl;
 773         }
 774     }
 775
 776 #ifdef WITH_MQ
 777     OCStackApplicationResult createMQTopicCallback(void* ctx, OCDoHandle /*handle*/,
 778                     OCClientResponse* clientResponse)
 779     {
 780         ClientCallbackContext::MQTopicContext* context =
 781             static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
 782         HeaderOptions serverHeaderOptions;
 783
 784         if (!clientResponse || !context)
 785         {
 786             return OC_STACK_DELETE_TRANSACTION;
 787         }
 788
 789         std::string createdUri;
 790         bool isLocationOption = false;
 791         OCStackResult result = clientResponse->result;
 792         if (OC_STACK_OK               == result ||
 793             OC_STACK_RESOURCE_CREATED == result)
 794         {
 795             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 796
 797             for (auto headerOption : serverHeaderOptions)
 798             {
 799                 if (HeaderOption::LOCATION_PATH_OPTION_ID == headerOption.getOptionID())
 800                 {
 801                     createdUri += "/";
 802                     createdUri += headerOption.getOptionData();
 803                     if (!isLocationOption)
 804                     {
 805                         isLocationOption = true;
 806                     }
 807                 }
 808             }
 809         }
 810
 811         if (!isLocationOption && NULL != clientResponse->resourceUri)
 812         {
 813             createdUri = std::string(clientResponse->resourceUri);
 814         }
 815
 816         auto clientWrapper = context->clientWrapper.lock();
 817
 818         if (!clientWrapper)
 819         {
 820             oclog() << "createMQTopicCallback(): failed to get a shared_ptr to the client wrapper"
 821                     << std::flush;
 822             return OC_STACK_DELETE_TRANSACTION;
 823         }
 824
 825         try
 826         {
 827             if (OC_STACK_OK               == result ||
 828                 OC_STACK_RESOURCE_CREATED == result)
 829             {
 830                 ListenOCContainer container(clientWrapper, clientResponse->devAddr,
 831                                             createdUri);
 832                 for (auto resource : container.Resources())
 833                 {
 834                     std::thread exec(context->callback, result,
 835                                      createdUri,
 836                                      resource);
 837                     exec.detach();
 838                 }
 839             }
 840             else
 841             {
 842                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 843                 std::thread exec(context->callback, result,
 844                                  createdUri,
 845                                  nullptr);
 846                 exec.detach();
 847             }
 848         }
 849         catch (std::exception &e)
 850         {
 851             oclog() << "Exception in createMQTopicCallback, ignoring response: "
 852                     << e.what() << std::flush;
 853         }
 854         return OC_STACK_DELETE_TRANSACTION;
 855     }
 856
 857     OCStackResult InProcClientWrapper::PutMQTopicRepresentation(
 858                 const OCDevAddr& devAddr,
 859                 const std::string& uri,
 860                 const OCRepresentation& rep,
 861                 const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 862                 MQTopicCallback& callback, QualityOfService QoS)
 863     {
 864         if (!callback)
 865         {
 866             return OC_STACK_INVALID_PARAM;
 867         }
 868         OCStackResult result;
 869         ClientCallbackContext::MQTopicContext* ctx =
 870                 new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
 871         OCCallbackData cbdata;
 872         cbdata.context = static_cast<void*>(ctx),
 873         cbdata.cb      = createMQTopicCallback;
 874         cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 875
 876         std::string url = assembleSetResourceUri(uri, queryParams);
 877
 878         auto cLock = m_csdkLock.lock();
 879
 880         if (cLock)
 881         {
 882             std::lock_guard<std::recursive_mutex> lock(*cLock);
 883             OCHeaderOption options[MAX_HEADER_OPTIONS];
 884
 885             result = OCDoResource(nullptr, OC_REST_PUT,
 886                                   url.c_str(), &devAddr,
 887                                   assembleSetResourcePayload(rep),
 888                                   CT_DEFAULT,
 889                                   static_cast<OCQualityOfService>(QoS),
 890                                   &cbdata,
 891                                   assembleHeaderOptions(options, headerOptions),
 892                                   headerOptions.size());
 893         }
 894         else
 895         {
 896             delete ctx;
 897             result = OC_STACK_ERROR;
 898         }
 899
 900         return result;
 901     }
 902 #endif
 903     OCStackApplicationResult getResourceCallback(void* ctx,
 904                                                  OCDoHandle /*handle*/,
 905         OCClientResponse* clientResponse)
 906     {
 907         ClientCallbackContext::GetContext* context =
 908             static_cast<ClientCallbackContext::GetContext*>(ctx);
 909         OCRepresentation rep;
 910         HeaderOptions serverHeaderOptions;
 911         OCStackResult result = clientResponse->result;
 912
 913         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 914         try
 915         {
 916             rep = parseGetSetCallback(clientResponse);
 917         }
 918         catch(OC::OCException& e)
 919         {
 920             result = e.code();
 921         }
 922
 923         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 924         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 925         exec.detach();
 926         return OC_STACK_DELETE_TRANSACTION;
 927     }
 928
 929     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 930         const OCDevAddr& devAddr,
 931         const std::string& resourceUri,
 932         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 933         OCConnectivityType connectivityType,
 934         GetCallback& callback, QualityOfService QoS)
 935     {
 936         if (!callback)
 937         {
 938             return OC_STACK_INVALID_PARAM;
 939         }
 940         OCStackResult result;
 941         ClientCallbackContext::GetContext* ctx =
 942             new ClientCallbackContext::GetContext(callback);
 943
 944         OCCallbackData cbdata;
 945         cbdata.context = static_cast<void*>(ctx);
 946         cbdata.cb      = getResourceCallback;
 947         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 948
 949         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 950
 951         auto cLock = m_csdkLock.lock();
 952
 953         if (cLock)
 954         {
 955             std::lock_guard<std::recursive_mutex> lock(*cLock);
 956             OCHeaderOption options[MAX_HEADER_OPTIONS];
 957
 958             result = OCDoResource(
 959                                   nullptr, OC_REST_GET,
 960                                   uri.c_str(),
 961                                   &devAddr, nullptr,
 962                                   connectivityType,
 963                                   static_cast<OCQualityOfService>(QoS),
 964                                   &cbdata,
 965                                   assembleHeaderOptions(options, headerOptions),
 966                                   headerOptions.size());
 967         }
 968         else
 969         {
 970             delete ctx;
 971             result = OC_STACK_ERROR;
 972         }
 973         return result;
 974     }
 975
 976
 977     OCStackApplicationResult setResourceCallback(void* ctx,
 978                                                  OCDoHandle /*handle*/,
 979         OCClientResponse* clientResponse)
 980     {
 981         ClientCallbackContext::SetContext* context =
 982             static_cast<ClientCallbackContext::SetContext*>(ctx);
 983         OCRepresentation attrs;
 984         HeaderOptions serverHeaderOptions;
 985
 986         OCStackResult result = clientResponse->result;
 987
 988         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
 989         try
 990         {
 991             attrs = parseGetSetCallback(clientResponse);
 992         }
 993         catch(OC::OCException& e)
 994         {
 995             result = e.code();
 996         }
 997
 998         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 999         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
1000         exec.detach();
1001         return OC_STACK_DELETE_TRANSACTION;
1002     }
1003
1004     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1005         const QueryParamsMap& queryParams)
1006     {
1007         if (!uri.empty())
1008         {
1009             if (uri.back() == '/')
1010             {
1011                 uri.resize(uri.size() - 1);
1012             }
1013         }
1014
1015         ostringstream paramsList;
1016         if (queryParams.size() > 0)
1017         {
1018             paramsList << '?';
1019         }
1020
1021         for (auto& param : queryParams)
1022         {
1023             paramsList << param.first <<'='<<param.second<<';';
1024         }
1025
1026         std::string queryString = paramsList.str();
1027
1028         if (queryString.empty())
1029         {
1030             return uri;
1031         }
1032
1033         if (queryString.back() == ';')
1034         {
1035             queryString.resize(queryString.size() - 1);
1036         }
1037
1038         std::string ret = uri + queryString;
1039         return ret;
1040     }
1041
1042     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1043         const QueryParamsList& queryParams)
1044     {
1045         if (!uri.empty())
1046         {
1047             if (uri.back() == '/')
1048             {
1049                 uri.resize(uri.size() - 1);
1050             }
1051         }
1052
1053         ostringstream paramsList;
1054         if (queryParams.size() > 0)
1055         {
1056             paramsList << '?';
1057         }
1058
1059         for (auto& param : queryParams)
1060         {
1061             for (auto& paramList : param.second)
1062             {
1063                 paramsList << param.first << '=' << paramList << ';';
1064             }
1065         }
1066
1067         std::string queryString = paramsList.str();
1068
1069         if (queryString.empty())
1070         {
1071             return uri;
1072         }
1073
1074         if (queryString.back() == ';')
1075         {
1076             queryString.resize(queryString.size() - 1);
1077         }
1078
1079         std::string ret = uri + queryString;
1080         return ret;
1081     }
1082
1083     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
1084     {
1085         MessageContainer ocInfo;
1086         ocInfo.addRepresentation(rep);
1087         for(const OCRepresentation& r : rep.getChildren())
1088         {
1089             ocInfo.addRepresentation(r);
1090         }
1091
1092         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
1093     }
1094
1095     OCStackResult InProcClientWrapper::PostResourceRepresentation(
1096         const OCDevAddr& devAddr,
1097         const std::string& uri,
1098         const OCRepresentation& rep,
1099         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1100         OCConnectivityType connectivityType,
1101         PostCallback& callback, QualityOfService QoS)
1102     {
1103         if (!callback)
1104         {
1105             return OC_STACK_INVALID_PARAM;
1106         }
1107         OCStackResult result;
1108         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1109         OCCallbackData cbdata;
1110         cbdata.context = static_cast<void*>(ctx),
1111         cbdata.cb      = setResourceCallback;
1112         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1113
1114
1115         std::string url = assembleSetResourceUri(uri, queryParams);
1116
1117         auto cLock = m_csdkLock.lock();
1118
1119         if (cLock)
1120         {
1121             std::lock_guard<std::recursive_mutex> lock(*cLock);
1122             OCHeaderOption options[MAX_HEADER_OPTIONS];
1123
1124             result = OCDoResource(nullptr, OC_REST_POST,
1125                                   url.c_str(), &devAddr,
1126                                   assembleSetResourcePayload(rep),
1127                                   connectivityType,
1128                                   static_cast<OCQualityOfService>(QoS),
1129                                   &cbdata,
1130                                   assembleHeaderOptions(options, headerOptions),
1131                                   headerOptions.size());
1132         }
1133         else
1134         {
1135             delete ctx;
1136             result = OC_STACK_ERROR;
1137         }
1138
1139         return result;
1140     }
1141
1142     OCStackResult InProcClientWrapper::PutResourceRepresentation(
1143         const OCDevAddr& devAddr,
1144         const std::string& uri,
1145         const OCRepresentation& rep,
1146         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1147         PutCallback& callback, QualityOfService QoS)
1148     {
1149         if (!callback)
1150         {
1151             return OC_STACK_INVALID_PARAM;
1152         }
1153         OCStackResult result;
1154         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1155         OCCallbackData cbdata;
1156         cbdata.context = static_cast<void*>(ctx),
1157         cbdata.cb      = setResourceCallback;
1158         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1159
1160
1161         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1162
1163         auto cLock = m_csdkLock.lock();
1164
1165         if (cLock)
1166         {
1167             std::lock_guard<std::recursive_mutex> lock(*cLock);
1168             OCDoHandle handle;
1169             OCHeaderOption options[MAX_HEADER_OPTIONS];
1170
1171             result = OCDoResource(&handle, OC_REST_PUT,
1172                                   url.c_str(), &devAddr,
1173                                   assembleSetResourcePayload(rep),
1174                                   CT_DEFAULT,
1175                                   static_cast<OCQualityOfService>(QoS),
1176                                   &cbdata,
1177                                   assembleHeaderOptions(options, headerOptions),
1178                                   headerOptions.size());
1179         }
1180         else
1181         {
1182             delete ctx;
1183             result = OC_STACK_ERROR;
1184         }
1185
1186         return result;
1187     }
1188
1189     OCStackApplicationResult deleteResourceCallback(void* ctx,
1190                                                     OCDoHandle /*handle*/,
1191         OCClientResponse* clientResponse)
1192     {
1193         ClientCallbackContext::DeleteContext* context =
1194             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
1195         HeaderOptions serverHeaderOptions;
1196
1197         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1198
1199         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1200         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
1201         exec.detach();
1202         return OC_STACK_DELETE_TRANSACTION;
1203     }
1204
1205     OCStackResult InProcClientWrapper::DeleteResource(
1206         const OCDevAddr& devAddr,
1207         const std::string& uri,
1208         const HeaderOptions& headerOptions,
1209         OCConnectivityType connectivityType,
1210         DeleteCallback& callback,
1211         QualityOfService /*QoS*/)
1212     {
1213         if (!callback)
1214         {
1215             return OC_STACK_INVALID_PARAM;
1216         }
1217         OCStackResult result;
1218         ClientCallbackContext::DeleteContext* ctx =
1219             new ClientCallbackContext::DeleteContext(callback);
1220         OCCallbackData cbdata;
1221         cbdata.context = static_cast<void*>(ctx),
1222         cbdata.cb      = deleteResourceCallback;
1223         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
1224
1225
1226         auto cLock = m_csdkLock.lock();
1227
1228         if (cLock)
1229         {
1230             OCHeaderOption options[MAX_HEADER_OPTIONS];
1231
1232             std::lock_guard<std::recursive_mutex> lock(*cLock);
1233
1234             result = OCDoResource(nullptr, OC_REST_DELETE,
1235                                   uri.c_str(), &devAddr,
1236                                   nullptr,
1237                                   connectivityType,
1238                                   static_cast<OCQualityOfService>(m_cfg.QoS),
1239                                   &cbdata,
1240                                   assembleHeaderOptions(options, headerOptions),
1241                                   headerOptions.size());
1242         }
1243         else
1244         {
1245             delete ctx;
1246             result = OC_STACK_ERROR;
1247         }
1248
1249         return result;
1250     }
1251
1252     OCStackApplicationResult observeResourceCallback(void* ctx,
1253                                                      OCDoHandle /*handle*/,
1254         OCClientResponse* clientResponse)
1255     {
1256         ClientCallbackContext::ObserveContext* context =
1257             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
1258         OCRepresentation attrs;
1259         HeaderOptions serverHeaderOptions;
1260         uint32_t sequenceNumber = clientResponse->sequenceNumber;
1261         OCStackResult result = clientResponse->result;
1262
1263         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1264         try
1265         {
1266             attrs = parseGetSetCallback(clientResponse);
1267         }
1268         catch(OC::OCException& e)
1269         {
1270             result = e.code();
1271         }
1272
1273         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1274         std::thread exec(context->callback, serverHeaderOptions, attrs,
1275                     result, sequenceNumber);
1276         exec.detach();
1277         if (sequenceNumber == MAX_SEQUENCE_NUMBER + 1)
1278         {
1279             return OC_STACK_DELETE_TRANSACTION;
1280         }
1281
1282         return OC_STACK_KEEP_TRANSACTION;
1283     }
1284
1285     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
1286         const OCDevAddr& devAddr,
1287         const std::string& uri,
1288         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1289         ObserveCallback& callback, QualityOfService QoS)
1290     {
1291         if (!callback)
1292         {
1293             return OC_STACK_INVALID_PARAM;
1294         }
1295         OCStackResult result;
1296
1297         ClientCallbackContext::ObserveContext* ctx =
1298             new ClientCallbackContext::ObserveContext(callback);
1299         OCCallbackData cbdata;
1300         cbdata.context = static_cast<void*>(ctx),
1301         cbdata.cb      = observeResourceCallback;
1302         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1303
1304
1305         OCMethod method;
1306         if (observeType == ObserveType::Observe)
1307         {
1308             method = OC_REST_OBSERVE;
1309         }
1310         else if (observeType == ObserveType::ObserveAll)
1311         {
1312             method = OC_REST_OBSERVE_ALL;
1313         }
1314         else
1315         {
1316             method = OC_REST_OBSERVE_ALL;
1317         }
1318
1319         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1320
1321         auto cLock = m_csdkLock.lock();
1322
1323         if (cLock)
1324         {
1325             std::lock_guard<std::recursive_mutex> lock(*cLock);
1326             OCHeaderOption options[MAX_HEADER_OPTIONS];
1327
1328             result = OCDoResource(handle, method,
1329                                   url.c_str(), &devAddr,
1330                                   nullptr,
1331                                   CT_DEFAULT,
1332                                   static_cast<OCQualityOfService>(QoS),
1333                                   &cbdata,
1334                                   assembleHeaderOptions(options, headerOptions),
1335                                   headerOptions.size());
1336         }
1337         else
1338         {
1339             delete ctx;
1340             return OC_STACK_ERROR;
1341         }
1342
1343         return result;
1344     }
1345
1346     OCStackResult InProcClientWrapper::CancelObserveResource(
1347             OCDoHandle handle,
1348             const std::string& /*host*/,
1349             const std::string& /*uri*/,
1350             const HeaderOptions& headerOptions,
1351             QualityOfService QoS)
1352     {
1353         OCStackResult result;
1354         auto cLock = m_csdkLock.lock();
1355
1356         if (cLock)
1357         {
1358             std::lock_guard<std::recursive_mutex> lock(*cLock);
1359             OCHeaderOption options[MAX_HEADER_OPTIONS];
1360
1361             result = OCCancel(handle,
1362                     static_cast<OCQualityOfService>(QoS),
1363                     assembleHeaderOptions(options, headerOptions),
1364                     headerOptions.size());
1365         }
1366         else
1367         {
1368             result = OC_STACK_ERROR;
1369         }
1370
1371         return result;
1372     }
1373
1374     OCStackApplicationResult subscribePresenceCallback(void* ctx,
1375                                                        OCDoHandle /*handle*/,
1376             OCClientResponse* clientResponse)
1377     {
1378         ClientCallbackContext::SubscribePresenceContext* context =
1379         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
1380
1381         /*
1382          * This a hack while we rethink presence subscription.
1383          */
1384         std::string url = clientResponse->devAddr.addr;
1385
1386         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1387         std::thread exec(context->callback, clientResponse->result,
1388                     clientResponse->sequenceNumber, url);
1389
1390         exec.detach();
1391
1392         return OC_STACK_KEEP_TRANSACTION;
1393     }
1394
1395     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
1396         const std::string& host, const std::string& resourceType,
1397         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
1398     {
1399         if (!presenceHandler)
1400         {
1401             return OC_STACK_INVALID_PARAM;
1402         }
1403
1404         ClientCallbackContext::SubscribePresenceContext* ctx =
1405             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
1406         OCCallbackData cbdata;
1407         cbdata.context = static_cast<void*>(ctx),
1408         cbdata.cb      = subscribePresenceCallback;
1409         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
1410
1411
1412         auto cLock = m_csdkLock.lock();
1413
1414         std::ostringstream os;
1415         os << host << OC_RSRVD_PRESENCE_URI;
1416
1417         if (!resourceType.empty())
1418         {
1419             os << "?rt=" << resourceType;
1420         }
1421
1422         if (!cLock)
1423         {
1424             delete ctx;
1425             return OC_STACK_ERROR;
1426         }
1427
1428         return OCDoResource(handle, OC_REST_PRESENCE,
1429                             os.str().c_str(), nullptr,
1430                             nullptr, connectivityType,
1431                             OC_LOW_QOS, &cbdata, NULL, 0);
1432     }
1433
1434     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
1435     {
1436         OCStackResult result;
1437         auto cLock = m_csdkLock.lock();
1438
1439         if (cLock)
1440         {
1441             std::lock_guard<std::recursive_mutex> lock(*cLock);
1442             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
1443         }
1444         else
1445         {
1446             result = OC_STACK_ERROR;
1447         }
1448
1449         return result;
1450     }
1451
1452 #ifdef WITH_CLOUD
1453     OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
1454                                                                const std::string& host,
1455                                                                const std::vector<std::string>& di,
1456                                                                OCConnectivityType connectivityType,
1457                                                                ObserveCallback& callback)
1458     {
1459         if (!callback)
1460         {
1461             return OC_STACK_INVALID_PARAM;
1462         }
1463         OCStackResult result;
1464
1465         ClientCallbackContext::ObserveContext* ctx =
1466             new ClientCallbackContext::ObserveContext(callback);
1467         OCCallbackData cbdata;
1468         cbdata.context = static_cast<void*>(ctx),
1469         cbdata.cb      = observeResourceCallback;
1470         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1471
1472         auto cLock = m_csdkLock.lock();
1473
1474         if (cLock)
1475         {
1476             std::lock_guard<std::recursive_mutex> lock(*cLock);
1477
1478             std::ostringstream os;
1479             os << host << OC_RSRVD_DEVICE_PRESENCE_URI;
1480             QueryParamsList queryParams({{OC_RSRVD_DEVICE_ID, di}});
1481             std::string url = assembleSetResourceUri(os.str(), queryParams);
1482
1483             result = OCDoResource(handle, OC_REST_OBSERVE,
1484                                   url.c_str(), nullptr,
1485                                   nullptr, connectivityType,
1486                                   OC_LOW_QOS, &cbdata,
1487                                   nullptr, 0);
1488         }
1489         else
1490         {
1491             delete ctx;
1492             result = OC_STACK_ERROR;
1493         }
1494
1495         return result;
1496     }
1497 #endif
1498
1499     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
1500     {
1501         qos = m_cfg.QoS;
1502         return OC_STACK_OK;
1503     }
1504
1505     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
1506            const HeaderOptions& headerOptions)
1507     {
1508         int i = 0;
1509
1510         if ( headerOptions.size() == 0)
1511         {
1512             return nullptr;
1513         }
1514
1515         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
1516         {
1517             options[i] = OCHeaderOption();
1518             options[i].protocolID = OC_COAP_ID;
1519             options[i].optionID = it->getOptionID();
1520             options[i].optionLength = it->getOptionData().length() + 1;
1521             strncpy((char*)options[i].optionData, it->getOptionData().c_str(),
1522                 sizeof(options[i].optionLength) -1 );
1523             options[i].optionData[sizeof(options[i].optionLength) - 1] = 0;
1524             i++;
1525         }
1526
1527         return options;
1528     }
1529
1530     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
1531     {
1532         if (!dev)
1533         {
1534             return nullptr;
1535         }
1536
1537         OCDPDev_t* result = new OCDPDev_t(*dev);
1538         result->prm = new OCPrm_t[dev->prmLen];
1539         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
1540         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
1541     }
1542
1543     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
1544     {
1545         while(list)
1546         {
1547             dpList.push_back(cloneDevice(list));
1548             list = list->next;
1549         }
1550     }
1551
1552     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
1553             GetDirectPairedCallback& callback)
1554     {
1555         if (!callback || 0 == waittime)
1556         {
1557             return OC_STACK_INVALID_PARAM;
1558         }
1559
1560         OCStackResult result = OC_STACK_ERROR;
1561         const OCDPDev_t *list = nullptr;
1562         PairedDevices dpDeviceList;
1563
1564         auto cLock = m_csdkLock.lock();
1565
1566         if (cLock)
1567         {
1568             std::lock_guard<std::recursive_mutex> lock(*cLock);
1569
1570             list = OCDiscoverDirectPairingDevices(waittime);
1571             if (NULL == list)
1572             {
1573                 result = OC_STACK_NO_RESOURCE;
1574                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1575                     << std::flush;
1576             }
1577             else {
1578                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1579                 convert(list, dpDeviceList);
1580                 std::thread exec(callback, dpDeviceList);
1581                 exec.detach();
1582                 result = OC_STACK_OK;
1583             }
1584         }
1585         else
1586         {
1587             result = OC_STACK_ERROR;
1588         }
1589
1590         return result;
1591     }
1592
1593     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1594     {
1595         if (!callback)
1596         {
1597             return OC_STACK_INVALID_PARAM;
1598         }
1599
1600         OCStackResult result = OC_STACK_ERROR;
1601         const OCDPDev_t *list = nullptr;
1602         PairedDevices dpDeviceList;
1603
1604         auto cLock = m_csdkLock.lock();
1605
1606         if (cLock)
1607         {
1608             std::lock_guard<std::recursive_mutex> lock(*cLock);
1609
1610             list = OCGetDirectPairedDevices();
1611             if (NULL == list)
1612             {
1613                 result = OC_STACK_NO_RESOURCE;
1614                 OIC_LOG_V(DEBUG, TAG, "%s: No device found for direct pairing", __func__);
1615             }
1616             else {
1617                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1618                 convert(list, dpDeviceList);
1619                 std::thread exec(callback, dpDeviceList);
1620                 exec.detach();
1621                 result = OC_STACK_OK;
1622             }
1623         }
1624         else
1625         {
1626             result = OC_STACK_ERROR;
1627         }
1628
1629         return result;
1630     }
1631
1632     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1633             OCStackResult result)
1634     {
1635
1636         ClientCallbackContext::DirectPairingContext* context =
1637             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1638
1639         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1640         std::thread exec(context->callback, cloneDevice(peer), result);
1641         exec.detach();
1642     }
1643
1644     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1645             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1646     {
1647         if (!peer || !callback)
1648         {
1649             oclog() << "Invalid parameters" << std::flush;
1650             return OC_STACK_INVALID_PARAM;
1651         }
1652
1653         OCStackResult result = OC_STACK_ERROR;
1654         ClientCallbackContext::DirectPairingContext* context =
1655             new ClientCallbackContext::DirectPairingContext(callback);
1656
1657         auto cLock = m_csdkLock.lock();
1658         if (cLock)
1659         {
1660             std::lock_guard<std::recursive_mutex> lock(*cLock);
1661             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1662                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1663         }
1664         else
1665         {
1666             delete context;
1667             result = OC_STACK_ERROR;
1668         }
1669         return result;
1670     }
1671 #ifdef TCP_ADAPTER
1672     OCStackApplicationResult KeepAliveRespCallback(void* ctx,
1673                                                  OCDoHandle /*handle*/,
1674         OCClientResponse* clientResponse)
1675     {
1676         ClientCallbackContext::KeepAliveContext* context =
1677             static_cast<ClientCallbackContext::KeepAliveContext*>(ctx);
1678         OCRepresentation attrs;
1679         OCStackResult result = clientResponse->result;
1680
1681         try
1682         {
1683             attrs = parseGetSetCallback(clientResponse);
1684         }
1685         catch(OC::OCException& e)
1686         {
1687             result = e.code();
1688         }
1689
1690         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1691         std::thread exec(context->callback, result, attrs);
1692         exec.detach();
1693         return OC_STACK_DELETE_TRANSACTION;
1694     }
1695
1696     OCStackResult InProcClientWrapper::findKeepAliveResource(std::string host,
1697                                                              KeepAliveCallback resultCallback)
1698     {
1699         if (host.empty() || !resultCallback)
1700         {
1701             oclog() << "Invalid parameters" << std::flush;
1702             return OC_STACK_INVALID_PARAM;
1703         }
1704
1705         OCStackResult result = OC_STACK_ERROR;
1706
1707         ClientCallbackContext::KeepAliveContext* ctx =
1708                        new ClientCallbackContext::KeepAliveContext(resultCallback);
1709         OCCallbackData cbdata;
1710         cbdata.context = static_cast<void*>(ctx),
1711         cbdata.cb      = KeepAliveRespCallback;
1712         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1713
1714         auto cLock = m_csdkLock.lock();
1715
1716         if (cLock)
1717         {
1718             std::lock_guard<std::recursive_mutex> lock(*cLock);
1719             result = OCFindKeepAliveResource(nullptr, host.c_str(), &cbdata);
1720         }
1721         else
1722         {
1723             delete ctx;
1724             result = OC_STACK_ERROR;
1725         }
1726         return result;
1727     }
1728
1729     OCStackResult InProcClientWrapper::sendKeepAliveRequest(std::string host,
1730                                                             const OCRepresentation& rep,
1731                                                             KeepAliveCallback resultCallback)
1732     {
1733         if (!resultCallback)
1734         {
1735             oclog() << "Invalid parameters" << std::flush;
1736             return OC_STACK_INVALID_PARAM;
1737         }
1738
1739         OCStackResult result = OC_STACK_ERROR;
1740
1741         ClientCallbackContext::KeepAliveContext* ctx = new ClientCallbackContext::KeepAliveContext(resultCallback);
1742         OCCallbackData cbdata;
1743         cbdata.context = static_cast<void*>(ctx),
1744         cbdata.cb      = KeepAliveRespCallback;
1745         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1746
1747         auto cLock = m_csdkLock.lock();
1748
1749         if (cLock)
1750         {
1751             std::lock_guard<std::recursive_mutex> lock(*cLock);
1752             OCRepPayload *payload = rep.getPayload();
1753             result = OCSendKeepAliveRequest (nullptr, host.c_str(), (OCPayload*)payload, &cbdata);
1754             if (result != OC_STACK_OK)
1755             {
1756                 OCRepPayloadDestroy(payload);
1757             }
1758         }
1759         else
1760         {
1761             delete ctx;
1762             result = OC_STACK_ERROR;
1763         }
1764         return result;
1765     }
1766 #endif
1767 }