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         }
 966         catch(OC::OCException& e)
 967         {
 968             result = e.code();
 969         }
 970
 971         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
 972         std::thread exec(context->callback, serverHeaderOptions, rep, result);
 973         exec.detach();
 974         return OC_STACK_DELETE_TRANSACTION;
 975     }
 976
 977     OCStackResult InProcClientWrapper::GetResourceRepresentation(
 978         const OCDevAddr& devAddr,
 979         const std::string& resourceUri,
 980         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
 981         OCConnectivityType connectivityType,
 982         GetCallback& callback, QualityOfService QoS)
 983     {
 984         if (!callback)
 985         {
 986             return OC_STACK_INVALID_PARAM;
 987         }
 988
 989         if (headerOptions.size() > MAX_HEADER_OPTIONS)
 990         {
 991             oclog() << "GetResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
 992             return OC_STACK_INVALID_PARAM;
 993         }
 994
 995         OCStackResult result;
 996         ClientCallbackContext::GetContext* ctx =
 997             new ClientCallbackContext::GetContext(callback);
 998
 999         OCCallbackData cbdata;
1000         cbdata.context = static_cast<void*>(ctx);
1001         cbdata.cb      = getResourceCallback;
1002         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
1003
1004         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
1005
1006         auto cLock = m_csdkLock.lock();
1007
1008         if (cLock)
1009         {
1010             std::lock_guard<std::recursive_mutex> lock(*cLock);
1011             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1012
1013             result = OCDoResource(
1014                                   nullptr, OC_REST_GET,
1015                                   uri.c_str(),
1016                                   &devAddr, nullptr,
1017                                   connectivityType,
1018                                   static_cast<OCQualityOfService>(QoS),
1019                                   &cbdata,
1020                                   options,
1021                                   headerOptions.size());
1022             delete[] options;
1023         }
1024         else
1025         {
1026             delete ctx;
1027             result = OC_STACK_ERROR;
1028         }
1029         return result;
1030     }
1031
1032
1033     OCStackApplicationResult setResourceCallback(void* ctx,
1034                                                  OCDoHandle /*handle*/,
1035         OCClientResponse* clientResponse)
1036     {
1037         ClientCallbackContext::SetContext* context =
1038             static_cast<ClientCallbackContext::SetContext*>(ctx);
1039         OCRepresentation attrs;
1040         HeaderOptions serverHeaderOptions;
1041
1042         OCStackResult result = clientResponse->result;
1043
1044         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1045         try
1046         {
1047             attrs = parseGetSetCallback(clientResponse);
1048         }
1049         catch(OC::OCException& e)
1050         {
1051             result = e.code();
1052         }
1053
1054         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1055         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
1056         exec.detach();
1057         return OC_STACK_DELETE_TRANSACTION;
1058     }
1059
1060     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1061         const QueryParamsMap& queryParams)
1062     {
1063         if (!uri.empty())
1064         {
1065             if (uri.back() == '/')
1066             {
1067                 uri.resize(uri.size() - 1);
1068             }
1069         }
1070
1071         ostringstream paramsList;
1072         if (queryParams.size() > 0)
1073         {
1074             paramsList << '?';
1075         }
1076
1077         for (auto& param : queryParams)
1078         {
1079             paramsList << param.first <<'='<<param.second<<';';
1080         }
1081
1082         std::string queryString = paramsList.str();
1083
1084         if (queryString.empty())
1085         {
1086             return uri;
1087         }
1088
1089         if (queryString.back() == ';')
1090         {
1091             queryString.resize(queryString.size() - 1);
1092         }
1093
1094         std::string ret = uri + queryString;
1095         return ret;
1096     }
1097
1098     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
1099         const QueryParamsList& queryParams)
1100     {
1101         if (!uri.empty())
1102         {
1103             if (uri.back() == '/')
1104             {
1105                 uri.resize(uri.size() - 1);
1106             }
1107         }
1108
1109         ostringstream paramsList;
1110         if (queryParams.size() > 0)
1111         {
1112             paramsList << '?';
1113         }
1114
1115         for (auto& param : queryParams)
1116         {
1117             for (auto& paramList : param.second)
1118             {
1119                 paramsList << param.first << '=' << paramList << ';';
1120             }
1121         }
1122
1123         std::string queryString = paramsList.str();
1124
1125         if (queryString.empty())
1126         {
1127             return uri;
1128         }
1129
1130         if (queryString.back() == ';')
1131         {
1132             queryString.resize(queryString.size() - 1);
1133         }
1134
1135         std::string ret = uri + queryString;
1136         return ret;
1137     }
1138
1139     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
1140     {
1141         MessageContainer ocInfo;
1142         ocInfo.addRepresentation(rep);
1143         for(const OCRepresentation& r : rep.getChildren())
1144         {
1145             ocInfo.addRepresentation(r);
1146         }
1147
1148         return reinterpret_cast<OCPayload*>(ocInfo.getPayload());
1149     }
1150
1151     OCStackResult InProcClientWrapper::PostResourceRepresentation(
1152         const OCDevAddr& devAddr,
1153         const std::string& uri,
1154         const OCRepresentation& rep,
1155         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1156         OCConnectivityType connectivityType,
1157         PostCallback& callback, QualityOfService QoS)
1158     {
1159         if (!callback)
1160         {
1161             return OC_STACK_INVALID_PARAM;
1162         }
1163
1164         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1165         {
1166             oclog() << "PostResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1167             return OC_STACK_INVALID_PARAM;
1168         }
1169
1170         OCStackResult result;
1171         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1172         OCCallbackData cbdata;
1173         cbdata.context = static_cast<void*>(ctx),
1174         cbdata.cb      = setResourceCallback;
1175         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1176
1177
1178         std::string url = assembleSetResourceUri(uri, queryParams);
1179
1180         auto cLock = m_csdkLock.lock();
1181
1182         if (cLock)
1183         {
1184             std::lock_guard<std::recursive_mutex> lock(*cLock);
1185             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1186
1187             result = OCDoResource(nullptr, OC_REST_POST,
1188                                   url.c_str(), &devAddr,
1189                                   assembleSetResourcePayload(rep),
1190                                   connectivityType,
1191                                   static_cast<OCQualityOfService>(QoS),
1192                                   &cbdata,
1193                                   options,
1194                                   headerOptions.size());
1195             delete[] options;
1196         }
1197         else
1198         {
1199             delete ctx;
1200             result = OC_STACK_ERROR;
1201         }
1202
1203         return result;
1204     }
1205
1206     OCStackResult InProcClientWrapper::PutResourceRepresentation(
1207         const OCDevAddr& devAddr,
1208         const std::string& uri,
1209         const OCRepresentation& rep,
1210         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1211         PutCallback& callback, QualityOfService QoS)
1212     {
1213         if (!callback)
1214         {
1215             return OC_STACK_INVALID_PARAM;
1216         }
1217
1218         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1219         {
1220             oclog() << "PutResourceRepresentation: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1221             return OC_STACK_INVALID_PARAM;
1222         }
1223
1224         OCStackResult result;
1225         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
1226         OCCallbackData cbdata;
1227         cbdata.context = static_cast<void*>(ctx),
1228         cbdata.cb      = setResourceCallback;
1229         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
1230
1231
1232         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1233
1234         auto cLock = m_csdkLock.lock();
1235
1236         if (cLock)
1237         {
1238             std::lock_guard<std::recursive_mutex> lock(*cLock);
1239             OCDoHandle handle;
1240             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1241
1242             result = OCDoResource(&handle, OC_REST_PUT,
1243                                   url.c_str(), &devAddr,
1244                                   assembleSetResourcePayload(rep),
1245                                   CT_DEFAULT,
1246                                   static_cast<OCQualityOfService>(QoS),
1247                                   &cbdata,
1248                                   options,
1249                                   headerOptions.size());
1250             delete[] options;
1251         }
1252         else
1253         {
1254             delete ctx;
1255             result = OC_STACK_ERROR;
1256         }
1257
1258         return result;
1259     }
1260
1261     OCStackApplicationResult deleteResourceCallback(void* ctx,
1262                                                     OCDoHandle /*handle*/,
1263         OCClientResponse* clientResponse)
1264     {
1265         ClientCallbackContext::DeleteContext* context =
1266             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
1267         HeaderOptions serverHeaderOptions;
1268
1269         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1270
1271         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1272         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
1273         exec.detach();
1274         return OC_STACK_DELETE_TRANSACTION;
1275     }
1276
1277     OCStackResult InProcClientWrapper::DeleteResource(
1278         const OCDevAddr& devAddr,
1279         const std::string& uri,
1280         const HeaderOptions& headerOptions,
1281         OCConnectivityType connectivityType,
1282         DeleteCallback& callback,
1283         QualityOfService /*QoS*/)
1284     {
1285         if (!callback)
1286         {
1287             return OC_STACK_INVALID_PARAM;
1288         }
1289
1290         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1291         {
1292             oclog() << "DeleteResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1293             return OC_STACK_INVALID_PARAM;
1294         }
1295
1296         OCStackResult result;
1297         ClientCallbackContext::DeleteContext* ctx =
1298             new ClientCallbackContext::DeleteContext(callback);
1299         OCCallbackData cbdata;
1300         cbdata.context = static_cast<void*>(ctx),
1301         cbdata.cb      = deleteResourceCallback;
1302         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
1303
1304
1305         auto cLock = m_csdkLock.lock();
1306
1307         if (cLock)
1308         {
1309             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1310
1311             std::lock_guard<std::recursive_mutex> lock(*cLock);
1312
1313             result = OCDoResource(nullptr, OC_REST_DELETE,
1314                                   uri.c_str(), &devAddr,
1315                                   nullptr,
1316                                   connectivityType,
1317                                   static_cast<OCQualityOfService>(m_cfg.QoS),
1318                                   &cbdata,
1319                                   options,
1320                                   headerOptions.size());
1321             delete[] options;
1322         }
1323         else
1324         {
1325             delete ctx;
1326             result = OC_STACK_ERROR;
1327         }
1328
1329         return result;
1330     }
1331
1332     OCStackApplicationResult observeResourceCallback(void* ctx,
1333                                                      OCDoHandle /*handle*/,
1334         OCClientResponse* clientResponse)
1335     {
1336         ClientCallbackContext::ObserveContext* context =
1337             static_cast<ClientCallbackContext::ObserveContext*>(ctx);
1338         OCRepresentation attrs;
1339         HeaderOptions serverHeaderOptions;
1340         uint32_t sequenceNumber = clientResponse->sequenceNumber;
1341         OCStackResult result = clientResponse->result;
1342
1343         parseServerHeaderOptions(clientResponse, serverHeaderOptions);
1344         try
1345         {
1346             attrs = parseGetSetCallback(clientResponse);
1347         }
1348         catch(OC::OCException& e)
1349         {
1350             result = e.code();
1351         }
1352
1353         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1354         std::thread exec(context->callback, serverHeaderOptions, attrs,
1355                     result, sequenceNumber);
1356         exec.detach();
1357         if (sequenceNumber == MAX_SEQUENCE_NUMBER + 1)
1358         {
1359             return OC_STACK_DELETE_TRANSACTION;
1360         }
1361
1362         return OC_STACK_KEEP_TRANSACTION;
1363     }
1364
1365     OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
1366         const OCDevAddr& devAddr,
1367         const std::string& uri,
1368         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
1369         ObserveCallback& callback, QualityOfService QoS)
1370     {
1371         if (!callback)
1372         {
1373             return OC_STACK_INVALID_PARAM;
1374         }
1375
1376         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1377         {
1378             oclog() << "ObserveResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1379             return OC_STACK_INVALID_PARAM;
1380         }
1381
1382         OCStackResult result;
1383
1384         ClientCallbackContext::ObserveContext* ctx =
1385             new ClientCallbackContext::ObserveContext(callback);
1386         OCCallbackData cbdata;
1387         cbdata.context = static_cast<void*>(ctx),
1388         cbdata.cb      = observeResourceCallback;
1389         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1390
1391
1392         OCMethod method;
1393         if (observeType == ObserveType::Observe)
1394         {
1395             method = OC_REST_OBSERVE;
1396         }
1397         else if (observeType == ObserveType::ObserveAll)
1398         {
1399             method = OC_REST_OBSERVE_ALL;
1400         }
1401         else
1402         {
1403             method = OC_REST_OBSERVE_ALL;
1404         }
1405
1406         std::string url = assembleSetResourceUri(uri, queryParams).c_str();
1407
1408         auto cLock = m_csdkLock.lock();
1409
1410         if (cLock)
1411         {
1412             std::lock_guard<std::recursive_mutex> lock(*cLock);
1413             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1414
1415             result = OCDoResource(handle, method,
1416                                   url.c_str(), &devAddr,
1417                                   nullptr,
1418                                   CT_DEFAULT,
1419                                   static_cast<OCQualityOfService>(QoS),
1420                                   &cbdata,
1421                                   options,
1422                                   headerOptions.size());
1423             delete[] options;
1424         }
1425         else
1426         {
1427             delete ctx;
1428             return OC_STACK_ERROR;
1429         }
1430
1431         return result;
1432     }
1433
1434     OCStackResult InProcClientWrapper::CancelObserveResource(
1435             OCDoHandle handle,
1436             const std::string& /*host*/,
1437             const std::string& /*uri*/,
1438             const HeaderOptions& headerOptions,
1439             QualityOfService QoS)
1440     {
1441         if (headerOptions.size() > MAX_HEADER_OPTIONS)
1442         {
1443             oclog() << "CancelObserveResource: Header options are more than MAX_HEADER_OPTIONS" << std::flush;
1444             return OC_STACK_INVALID_PARAM;
1445         }
1446
1447         OCStackResult result;
1448         auto cLock = m_csdkLock.lock();
1449
1450         if (cLock)
1451         {
1452             std::lock_guard<std::recursive_mutex> lock(*cLock);
1453             OCHeaderOption *options = assembleHeaderOptions(headerOptions);
1454
1455             result = OCCancel(handle,
1456                     static_cast<OCQualityOfService>(QoS),
1457                     options,
1458                     headerOptions.size());
1459             delete[] options;
1460         }
1461         else
1462         {
1463             result = OC_STACK_ERROR;
1464         }
1465
1466         return result;
1467     }
1468
1469 #ifdef WITH_PRESENCE
1470     OCStackApplicationResult subscribePresenceCallback(void* ctx,
1471                                                        OCDoHandle /*handle*/,
1472             OCClientResponse* clientResponse)
1473     {
1474         ClientCallbackContext::SubscribePresenceContext* context =
1475         static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
1476
1477         /*
1478          * This a hack while we rethink presence subscription.
1479          */
1480         std::string url = clientResponse->devAddr.addr;
1481
1482         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1483         std::thread exec(context->callback, clientResponse->result,
1484                     clientResponse->sequenceNumber, url);
1485
1486         exec.detach();
1487
1488         return OC_STACK_KEEP_TRANSACTION;
1489     }
1490 #endif
1491
1492     OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
1493         const std::string& host, const std::string& resourceType,
1494         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
1495     {
1496 #ifdef WITH_PRESENCE
1497         if (!presenceHandler)
1498         {
1499             return OC_STACK_INVALID_PARAM;
1500         }
1501
1502         ClientCallbackContext::SubscribePresenceContext* ctx =
1503             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
1504         OCCallbackData cbdata;
1505         cbdata.context = static_cast<void*>(ctx),
1506         cbdata.cb      = subscribePresenceCallback;
1507         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
1508
1509
1510         auto cLock = m_csdkLock.lock();
1511
1512         std::ostringstream os;
1513         os << host << OC_RSRVD_PRESENCE_URI;
1514
1515         if (!resourceType.empty())
1516         {
1517             os << "?rt=" << resourceType;
1518         }
1519
1520         if (!cLock)
1521         {
1522             delete ctx;
1523             return OC_STACK_ERROR;
1524         }
1525
1526         return OCDoResource(handle, OC_REST_PRESENCE,
1527                             os.str().c_str(), nullptr,
1528                             nullptr, connectivityType,
1529                             OC_LOW_QOS, &cbdata, NULL, 0);
1530 #else
1531         return OC_STACK_NOT_IMPLEMENTED;
1532 #endif
1533     }
1534
1535     OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
1536     {
1537 #ifdef WITH_PRESENCE
1538         OCStackResult result;
1539         auto cLock = m_csdkLock.lock();
1540
1541         if (cLock)
1542         {
1543             std::lock_guard<std::recursive_mutex> lock(*cLock);
1544             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
1545         }
1546         else
1547         {
1548             result = OC_STACK_ERROR;
1549         }
1550
1551         return result;
1552 #else
1553         return OC_STACK_NOT_IMPLEMENTED;
1554 #endif
1555     }
1556
1557 #ifdef WITH_CLOUD
1558     OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
1559                                                                const std::string& host,
1560                                                                const std::vector<std::string>& di,
1561                                                                OCConnectivityType connectivityType,
1562                                                                ObserveCallback& callback)
1563     {
1564         if (!callback)
1565         {
1566             return OC_STACK_INVALID_PARAM;
1567         }
1568         OCStackResult result;
1569
1570         ClientCallbackContext::ObserveContext* ctx =
1571             new ClientCallbackContext::ObserveContext(callback);
1572         OCCallbackData cbdata;
1573         cbdata.context = static_cast<void*>(ctx),
1574         cbdata.cb      = observeResourceCallback;
1575         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
1576
1577         auto cLock = m_csdkLock.lock();
1578
1579         if (cLock)
1580         {
1581             std::lock_guard<std::recursive_mutex> lock(*cLock);
1582
1583             std::ostringstream os;
1584             os << host << OC_RSRVD_DEVICE_PRESENCE_URI;
1585             QueryParamsList queryParams({{OC_RSRVD_DEVICE_ID, di}});
1586             std::string url = assembleSetResourceUri(os.str(), queryParams);
1587
1588             result = OCDoResource(handle, OC_REST_OBSERVE,
1589                                   url.c_str(), nullptr,
1590                                   nullptr, connectivityType,
1591                                   OC_LOW_QOS, &cbdata,
1592                                   nullptr, 0);
1593         }
1594         else
1595         {
1596             delete ctx;
1597             result = OC_STACK_ERROR;
1598         }
1599
1600         return result;
1601     }
1602 #endif
1603
1604     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
1605     {
1606         qos = m_cfg.QoS;
1607         return OC_STACK_OK;
1608     }
1609
1610     OCHeaderOption* InProcClientWrapper::assembleHeaderOptions(const HeaderOptions& headerOptions)
1611     {
1612         if ( headerOptions.size() == 0)
1613         {
1614             return nullptr;
1615         }
1616
1617         OCHeaderOption* options = new OCHeaderOption[headerOptions.size()]();
1618
1619         size_t numOptions = 0;
1620         for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
1621         {
1622             OCStackResult ret = OCSetHeaderOption(options, &numOptions, it->getOptionID(),
1623                                     it->getOptionData().c_str(), it->getOptionData().length());
1624             if (OC_STACK_OK != ret)
1625             {
1626                 OIC_LOG_V(ERROR, TAG, "Failed to convert vnd header options! (error=%d)", ret);
1627                 delete[] options;
1628                 return nullptr;
1629             }
1630         }
1631
1632         return options;
1633     }
1634
1635     std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
1636     {
1637         if (!dev)
1638         {
1639             return nullptr;
1640         }
1641
1642         OCDPDev_t* result = new OCDPDev_t(*dev);
1643         result->prm = new OCPrm_t[dev->prmLen];
1644         memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
1645         return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
1646     }
1647
1648     void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
1649     {
1650         while(list)
1651         {
1652             dpList.push_back(cloneDevice(list));
1653             list = list->next;
1654         }
1655     }
1656
1657     OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
1658             GetDirectPairedCallback& callback)
1659     {
1660         if (!callback || 0 == waittime)
1661         {
1662             return OC_STACK_INVALID_PARAM;
1663         }
1664
1665         OCStackResult result = OC_STACK_ERROR;
1666         const OCDPDev_t *list = nullptr;
1667         PairedDevices dpDeviceList;
1668
1669         auto cLock = m_csdkLock.lock();
1670
1671         if (cLock)
1672         {
1673             std::lock_guard<std::recursive_mutex> lock(*cLock);
1674
1675             list = OCDiscoverDirectPairingDevices(waittime);
1676             if (NULL == list)
1677             {
1678                 result = OC_STACK_NO_RESOURCE;
1679                 oclog() << "findDirectPairingDevices(): No device found for direct pairing"
1680                     << std::flush;
1681             }
1682             else {
1683                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1684                 convert(list, dpDeviceList);
1685                 std::thread exec(callback, dpDeviceList);
1686                 exec.detach();
1687                 result = OC_STACK_OK;
1688             }
1689         }
1690         else
1691         {
1692             result = OC_STACK_ERROR;
1693         }
1694
1695         return result;
1696     }
1697
1698     OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
1699     {
1700         if (!callback)
1701         {
1702             return OC_STACK_INVALID_PARAM;
1703         }
1704
1705         OCStackResult result = OC_STACK_ERROR;
1706         const OCDPDev_t *list = nullptr;
1707         PairedDevices dpDeviceList;
1708
1709         auto cLock = m_csdkLock.lock();
1710
1711         if (cLock)
1712         {
1713             std::lock_guard<std::recursive_mutex> lock(*cLock);
1714
1715             list = OCGetDirectPairedDevices();
1716             if (NULL == list)
1717             {
1718                 result = OC_STACK_NO_RESOURCE;
1719                 OIC_LOG_V(DEBUG, TAG, "%s: No device found for direct pairing", __func__);
1720             }
1721             else {
1722                 OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1723                 convert(list, dpDeviceList);
1724                 std::thread exec(callback, dpDeviceList);
1725                 exec.detach();
1726                 result = OC_STACK_OK;
1727             }
1728         }
1729         else
1730         {
1731             result = OC_STACK_ERROR;
1732         }
1733
1734         return result;
1735     }
1736
1737     void directPairingCallback(void *ctx, OCDPDev_t *peer,
1738             OCStackResult result)
1739     {
1740
1741         ClientCallbackContext::DirectPairingContext* context =
1742             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
1743
1744         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1745         std::thread exec(context->callback, cloneDevice(peer), result);
1746         exec.detach();
1747     }
1748
1749     OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
1750             const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
1751     {
1752         if (!peer || !callback)
1753         {
1754             oclog() << "Invalid parameters" << std::flush;
1755             return OC_STACK_INVALID_PARAM;
1756         }
1757
1758         OCStackResult result = OC_STACK_ERROR;
1759         ClientCallbackContext::DirectPairingContext* context =
1760             new ClientCallbackContext::DirectPairingContext(callback);
1761
1762         auto cLock = m_csdkLock.lock();
1763         if (cLock)
1764         {
1765             std::lock_guard<std::recursive_mutex> lock(*cLock);
1766             result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
1767                     pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
1768         }
1769         else
1770         {
1771             delete context;
1772             result = OC_STACK_ERROR;
1773         }
1774         return result;
1775     }
1776 #ifdef TCP_ADAPTER
1777     OCStackApplicationResult KeepAliveRespCallback(void* ctx,
1778                                                  OCDoHandle /*handle*/,
1779         OCClientResponse* clientResponse)
1780     {
1781         ClientCallbackContext::KeepAliveContext* context =
1782             static_cast<ClientCallbackContext::KeepAliveContext*>(ctx);
1783         OCRepresentation attrs;
1784         OCStackResult result = clientResponse->result;
1785
1786         try
1787         {
1788             attrs = parseGetSetCallback(clientResponse);
1789         }
1790         catch(OC::OCException& e)
1791         {
1792             result = e.code();
1793         }
1794
1795         OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
1796         std::thread exec(context->callback, result, attrs);
1797         exec.detach();
1798         return OC_STACK_DELETE_TRANSACTION;
1799     }
1800
1801     OCStackResult InProcClientWrapper::findKeepAliveResource(std::string host,
1802                                                              KeepAliveCallback resultCallback)
1803     {
1804         if (host.empty() || !resultCallback)
1805         {
1806             oclog() << "Invalid parameters" << std::flush;
1807             return OC_STACK_INVALID_PARAM;
1808         }
1809
1810         OCStackResult result = OC_STACK_ERROR;
1811
1812         ClientCallbackContext::KeepAliveContext* ctx =
1813                        new ClientCallbackContext::KeepAliveContext(resultCallback);
1814         OCCallbackData cbdata;
1815         cbdata.context = static_cast<void*>(ctx),
1816         cbdata.cb      = KeepAliveRespCallback;
1817         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1818
1819         auto cLock = m_csdkLock.lock();
1820
1821         if (cLock)
1822         {
1823             std::lock_guard<std::recursive_mutex> lock(*cLock);
1824             result = OCFindKeepAliveResource(nullptr, host.c_str(), &cbdata);
1825         }
1826         else
1827         {
1828             delete ctx;
1829             result = OC_STACK_ERROR;
1830         }
1831         return result;
1832     }
1833
1834     OCStackResult InProcClientWrapper::sendKeepAliveRequest(std::string host,
1835                                                             const OCRepresentation& rep,
1836                                                             KeepAliveCallback resultCallback)
1837     {
1838         if (!resultCallback)
1839         {
1840             oclog() << "Invalid parameters" << std::flush;
1841             return OC_STACK_INVALID_PARAM;
1842         }
1843
1844         OCStackResult result = OC_STACK_ERROR;
1845
1846         ClientCallbackContext::KeepAliveContext* ctx = new ClientCallbackContext::KeepAliveContext(resultCallback);
1847         OCCallbackData cbdata;
1848         cbdata.context = static_cast<void*>(ctx),
1849         cbdata.cb      = KeepAliveRespCallback;
1850         cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
1851
1852         auto cLock = m_csdkLock.lock();
1853
1854         if (cLock)
1855         {
1856             std::lock_guard<std::recursive_mutex> lock(*cLock);
1857             OCRepPayload *payload = rep.getPayload();
1858             result = OCSendKeepAliveRequest (nullptr, host.c_str(), (OCPayload*)payload, &cbdata);
1859         }
1860         else
1861         {
1862             delete ctx;
1863             result = OC_STACK_ERROR;
1864         }
1865         return result;
1866     }
1867 #endif
1868 }