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