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