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