X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;f=resource%2Fsrc%2FInProcClientWrapper.cpp;h=abea511e674978a6ecdf61db0d60b50d8c2327f4;hb=7f00f942c39b7bc27c7eeecf213a239c3fe4173c;hp=1d68a2ff8af3261eeac480a2294fd8e1df8d15bd;hpb=112597ac33022a4ad678f228e97cdff627eee9b6;p=platform%2Fupstream%2Fiotivity.git

diff --git a/resource/src/InProcClientWrapper.cpp b/resource/src/InProcClientWrapper.cpp
index 1d68a2f..abea511 100644
--- a/resource/src/InProcClientWrapper.cpp
+++ b/resource/src/InProcClientWrapper.cpp
@@ -25,6 +25,13 @@
 #include "OCResource.h"
 #include "ocpayload.h"
 #include <OCSerialization.h>
+#include "logger.h"
+#ifdef TCP_ADAPTER
+#include "oickeepalive.h"
+#endif
+
+#define TAG "OIC_CLIENT_WRAPPER"
+
 using namespace std;
 
 namespace OC
@@ -36,6 +43,24 @@ namespace OC
     {
         // if the config type is server, we ought to never get called.  If the config type
         // is both, we count on the server to run the thread and do the initialize
+        start();
+    }
+
+    InProcClientWrapper::~InProcClientWrapper()
+    {
+        try
+        {
+            stop();
+        }
+        catch (InitializeException &e)
+        {
+            oclog() << "Exception in stop"<< e.what() << std::flush;
+        }
+    }
+
+    OCStackResult InProcClientWrapper::start()
+    {
+        OIC_LOG_V(INFO, TAG, "start ocplatform for client : %d", m_cfg.transportType);
 
         if (m_cfg.mode == ModeType::Client)
         {
@@ -43,32 +68,45 @@ namespace OC
                             static_cast<OCTransportFlags>(m_cfg.serverConnectivity & CT_MASK_FLAGS);
             OCTransportFlags clientFlags =
                             static_cast<OCTransportFlags>(m_cfg.clientConnectivity & CT_MASK_FLAGS);
-            OCStackResult result = OCInit1(OC_CLIENT, serverFlags, clientFlags);
+            OCStackResult result = OCInit2(OC_CLIENT, serverFlags, clientFlags,
+                                           m_cfg.transportType);
 
             if (OC_STACK_OK != result)
             {
                 throw InitializeException(OC::InitException::STACK_INIT_ERROR, result);
             }
 
-            m_threadRun = true;
-            m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
+            if (false == m_threadRun)
+            {
+                m_threadRun = true;
+                m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
+            }
         }
+        return OC_STACK_OK;
     }
 
-    InProcClientWrapper::~InProcClientWrapper()
+    OCStackResult InProcClientWrapper::stop()
     {
+        OIC_LOG(INFO, TAG, "stop ocplatform");
+
         if (m_threadRun && m_listeningThread.joinable())
         {
             m_threadRun = false;
             m_listeningThread.join();
         }
 
-        // only stop if we are the ones who actually called 'init'.  We are counting
+        // only stop if we are the ones who actually called 'start'.  We are counting
         // on the server to do the stop.
         if (m_cfg.mode == ModeType::Client)
         {
-            OCStop();
+            OCStackResult result = OCStop();
+
+            if (OC_STACK_OK != result)
+            {
+               throw InitializeException(OC::InitException::STACK_TERMINATE_ERROR, result);
+            }
         }
+        return OC_STACK_OK;
     }
 
     void InProcClientWrapper::listeningFunc()
@@ -101,19 +139,15 @@ namespace OC
     {
         if (clientResponse->payload == nullptr ||
                 (
-                    clientResponse->payload->type != PAYLOAD_TYPE_DEVICE &&
-                    clientResponse->payload->type != PAYLOAD_TYPE_PLATFORM &&
                     clientResponse->payload->type != PAYLOAD_TYPE_REPRESENTATION
                 )
           )
         {
-            //OCPayloadDestroy(clientResponse->payload);
             return OCRepresentation();
         }
 
         MessageContainer oc;
         oc.setPayload(clientResponse->payload);
-        //OCPayloadDestroy(clientResponse->payload);
 
         std::vector<OCRepresentation>::const_iterator it = oc.representations().begin();
         if (it == oc.representations().end())
@@ -122,21 +156,25 @@ namespace OC
         }
 
         // first one is considered the root, everything else is considered a child of this one.
-        OCRepresentation root = *it;
-        root.setDevAddr(clientResponse->devAddr);
-        root.setUri(clientResponse->resourceUri);
-        ++it;
+       OCRepresentation root = *it;
+       root.setDevAddr(clientResponse->devAddr);
+       root.setUri(clientResponse->resourceUri);
+       ++it;
 
         std::for_each(it, oc.representations().end(),
                 [&root](const OCRepresentation& repItr)
                 {root.addChild(repItr);});
         return root;
-
     }
 
     OCStackApplicationResult listenCallback(void* ctx, OCDoHandle /*handle*/,
         OCClientResponse* clientResponse)
     {
+        if (!ctx || !clientResponse)
+        {
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
         ClientCallbackContext::ListenContext* context =
             static_cast<ClientCallbackContext::ListenContext*>(ctx);
 
@@ -165,17 +203,20 @@ namespace OC
             return OC_STACK_KEEP_TRANSACTION;
         }
 
-        try{
+        try
+        {
             ListenOCContainer container(clientWrapper, clientResponse->devAddr,
                                     reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
             // loop to ensure valid construction of all resources
+
             for(auto resource : container.Resources())
             {
                 std::thread exec(context->callback, resource);
                 exec.detach();
             }
         }
-        catch (std::exception &e){
+        catch (std::exception &e)
+        {
             oclog() << "Exception in listCallback, ignoring response: "
                     << e.what() << std::flush;
         }
@@ -194,18 +235,14 @@ namespace OC
 
         ClientCallbackContext::ListenErrorContext* context =
             static_cast<ClientCallbackContext::ListenErrorContext*>(ctx);
-        if (!context)
-        {
-            return OC_STACK_KEEP_TRANSACTION;
-        }
 
         OCStackResult result = clientResponse->result;
         if (result == OC_STACK_OK)
         {
             if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
             {
-                oclog() << "listenCallback(): clientResponse payload was null or the wrong type"
-                    << std::flush;
+                OIC_LOG_V(DEBUG, TAG, "%s: clientResponse payload was null or the wrong type",
+                          __func__);
                 return OC_STACK_KEEP_TRANSACTION;
             }
 
@@ -213,8 +250,8 @@ namespace OC
 
             if (!clientWrapper)
             {
-                oclog() << "listenCallback(): failed to get a shared_ptr to the client wrapper"
-                        << std::flush;
+                OIC_LOG_V(DEBUG, TAG, "%s: failed to get a shared_ptr to the client wrapper",
+                          __func__);
                 return OC_STACK_KEEP_TRANSACTION;
             }
 
@@ -229,10 +266,16 @@ namespace OC
             return OC_STACK_KEEP_TRANSACTION;
         }
 
-        std::string resourceURI = clientResponse->resourceUri;
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
+        std::string resourceURI;
+        if(NULL != clientResponse->resourceUri)
+        {
+            resourceURI = clientResponse->resourceUri;
+        }
+
         std::thread exec(context->errorCallback, resourceURI, result);
         exec.detach();
-        return OC_STACK_DELETE_TRANSACTION;
+        return OC_STACK_KEEP_TRANSACTION;
     }
 
     OCStackResult InProcClientWrapper::ListenForResource(
@@ -325,6 +368,320 @@ namespace OC
         return result;
     }
 
+    OCStackApplicationResult listenResListCallback(void* ctx, OCDoHandle /*handle*/,
+        OCClientResponse* clientResponse)
+    {
+        if (!ctx || !clientResponse)
+        {
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        ClientCallbackContext::ListenResListContext* context =
+            static_cast<ClientCallbackContext::ListenResListContext*>(ctx);
+
+        if (clientResponse->result != OC_STACK_OK)
+        {
+            oclog() << "listenResListCallback(): failed to create resource. clientResponse: "
+                    << clientResponse->result
+                    << std::flush;
+
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
+        {
+            oclog() << "listenResListCallback(): clientResponse payload was null or the wrong type"
+                << std::flush;
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        auto clientWrapper = context->clientWrapper.lock();
+
+        if (!clientWrapper)
+        {
+            oclog() << "listenResListCallback(): failed to get a shared_ptr to the client wrapper"
+                    << std::flush;
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        try
+        {
+            ListenOCContainer container(clientWrapper, clientResponse->devAddr,
+                                    reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
+
+            OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
+            std::thread exec(context->callback, container.Resources());
+            exec.detach();
+        }
+        catch (std::exception &e)
+        {
+            oclog() << "Exception in listenResListCallback(), ignoring response: "
+                    << e.what() << std::flush;
+        }
+
+        return OC_STACK_KEEP_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::ListenForResourceList(
+            const std::string& serviceUrl,
+            const std::string& resourceType,
+            OCConnectivityType connectivityType,
+            FindResListCallback& callback, QualityOfService QoS)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result;
+        ostringstream resourceUri;
+        resourceUri << serviceUrl << resourceType;
+
+        ClientCallbackContext::ListenResListContext* context =
+            new ClientCallbackContext::ListenResListContext(callback, shared_from_this());
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(context),
+        cbdata.cb      = listenResListCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenResListContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            result = OCDoResource(nullptr, OC_REST_DISCOVER,
+                                  resourceUri.str().c_str(),
+                                  nullptr, nullptr, connectivityType,
+                                  static_cast<OCQualityOfService>(QoS),
+                                  &cbdata,
+                                  nullptr, 0);
+        }
+        else
+        {
+            delete context;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
+
+    OCStackApplicationResult listenResListWithErrorCallback(void* ctx, OCDoHandle /*handle*/,
+        OCClientResponse* clientResponse)
+    {
+        if (!ctx || !clientResponse)
+        {
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        ClientCallbackContext::ListenResListWithErrorContext* context =
+                static_cast<ClientCallbackContext::ListenResListWithErrorContext*>(ctx);
+
+        OCStackResult result = clientResponse->result;
+        if (result != OC_STACK_OK)
+        {
+            oclog() << "listenResListWithErrorCallback(): failed to create resource. clientResponse: "
+                    << result << std::flush;
+
+            //send the error callback
+            std::string uri;
+            if(NULL != clientResponse->resourceUri)
+            {
+                uri = clientResponse->resourceUri;
+            }
+            std::thread exec(context->errorCallback, uri, result);
+            exec.detach();
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
+        {
+            oclog() << "listenResListWithErrorCallback(): clientResponse payload was null or the wrong type"
+                << std::flush;
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        auto clientWrapper = context->clientWrapper.lock();
+
+        if (!clientWrapper)
+        {
+            oclog() << "listenResListWithErrorCallback(): failed to get a shared_ptr to the client wrapper"
+                    << std::flush;
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        try
+        {
+            ListenOCContainer container(clientWrapper, clientResponse->devAddr,
+                            reinterpret_cast<OCDiscoveryPayload*>(clientResponse->payload));
+
+            OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
+            std::thread exec(context->callback, container.Resources());
+            exec.detach();
+        }
+        catch (std::exception &e)
+        {
+            oclog() << "Exception in listenResListWithErrorCallback(), ignoring response: "
+            << e.what() << std::flush;
+        }
+
+        return OC_STACK_KEEP_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::ListenForResourceListWithError(
+            const std::string& serviceUrl,
+            const std::string& resourceType,
+            OCConnectivityType connectivityType,
+            FindResListCallback& callback,
+            FindErrorCallback& errorCallback, QualityOfService QoS)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result;
+        ostringstream resourceUri;
+        resourceUri << serviceUrl << resourceType;
+
+        ClientCallbackContext::ListenResListWithErrorContext* context =
+            new ClientCallbackContext::ListenResListWithErrorContext(callback, errorCallback,
+                                                          shared_from_this());
+        if (!context)
+        {
+            return OC_STACK_ERROR;
+        }
+
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(context),
+        cbdata.cb      = listenResListWithErrorCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenResListWithErrorContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            result = OCDoResource(nullptr, OC_REST_DISCOVER,
+                                  resourceUri.str().c_str(),
+                                  nullptr, nullptr, connectivityType,
+                                  static_cast<OCQualityOfService>(QoS),
+                                  &cbdata,
+                                  nullptr, 0);
+        }
+        else
+        {
+            delete context;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
+
+#ifdef WITH_MQ
+    OCStackApplicationResult listenMQCallback(void* ctx, OCDoHandle /*handle*/,
+                                              OCClientResponse* clientResponse)
+    {
+        ClientCallbackContext::MQTopicContext* context =
+            static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
+
+        if (!clientResponse || !context)
+        {
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        std::string resourceURI;
+        if(NULL != clientResponse->resourceUri)
+        {
+            resourceURI  = clientResponse->resourceUri;
+        }
+
+        if (clientResponse->result != OC_STACK_OK)
+        {
+            oclog() << "listenMQCallback(): failed to create resource. clientResponse: "
+                    << clientResponse->result
+                    << std::flush;
+
+            std::thread exec(context->callback, clientResponse->result,
+                             resourceURI, nullptr);
+            exec.detach();
+
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        auto clientWrapper = context->clientWrapper.lock();
+        if (!clientWrapper)
+        {
+            oclog() << "listenMQCallback(): failed to get a shared_ptr to the client wrapper"
+                    << std::flush;
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        try
+        {
+            ListenOCContainer container(clientWrapper, clientResponse->devAddr,
+                                        (OCRepPayload *) clientResponse->payload);
+
+            // loop to ensure valid construction of all resources
+            for (auto resource : container.Resources())
+            {
+                std::thread exec(context->callback, clientResponse->result,
+                                 resourceURI, resource);
+                exec.detach();
+            }
+        }
+        catch (std::exception &e)
+        {
+            oclog() << "Exception in listCallback, ignoring response: "
+                    << e.what() << std::flush;
+        }
+
+        return OC_STACK_DELETE_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::ListenForMQTopic(const OCDevAddr& devAddr,
+                                                        const std::string& resourceUri,
+                                                        const QueryParamsMap& queryParams,
+                                                        const HeaderOptions& headerOptions,
+                                                        MQTopicCallback& callback,
+                                                        QualityOfService QoS)
+    {
+        oclog() << "ListenForMQTopic()" << std::flush;
+
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        ClientCallbackContext::MQTopicContext* context =
+            new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(context),
+        cbdata.cb      = listenMQCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
+
+        std::string uri = assembleSetResourceUri(resourceUri, queryParams);
+
+        OCStackResult result = OC_STACK_ERROR;
+        auto cLock = m_csdkLock.lock();
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            OCHeaderOption options[MAX_HEADER_OPTIONS];
+            result = OCDoResource(
+                                  nullptr, OC_REST_GET,
+                                  uri.c_str(),
+                                  &devAddr, nullptr,
+                                  CT_DEFAULT,
+                                  static_cast<OCQualityOfService>(QoS),
+                                  &cbdata,
+                                  assembleHeaderOptions(options, headerOptions),
+                                  headerOptions.size());
+        }
+        else
+        {
+            delete context;
+        }
+
+        return result;
+    }
+#endif
+
     OCStackApplicationResult listenDeviceCallback(void* ctx,
                                                   OCDoHandle /*handle*/,
             OCClientResponse* clientResponse)
@@ -334,110 +691,236 @@ namespace OC
 
         try
         {
+            OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
             OCRepresentation rep = parseGetSetCallback(clientResponse);
             std::thread exec(context->callback, rep);
             exec.detach();
         }
-        catch(OC::OCException& e)
+        catch(OC::OCException& e)
+        {
+            oclog() <<"Exception in listenDeviceCallback, ignoring response: "
+                <<e.what() <<std::flush;
+        }
+
+        return OC_STACK_KEEP_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::ListenForDevice(
+            const std::string& serviceUrl,
+            const std::string& deviceURI,
+            OCConnectivityType connectivityType,
+            FindDeviceCallback& callback,
+            QualityOfService QoS)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+        OCStackResult result;
+        ostringstream deviceUri;
+        deviceUri << serviceUrl << deviceURI;
+
+        ClientCallbackContext::DeviceListenContext* context =
+            new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
+        OCCallbackData cbdata;
+
+        cbdata.context = static_cast<void*>(context),
+        cbdata.cb      = listenDeviceCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            result = OCDoResource(nullptr, OC_REST_DISCOVER,
+                                  deviceUri.str().c_str(),
+                                  nullptr, nullptr, connectivityType,
+                                  static_cast<OCQualityOfService>(QoS),
+                                  &cbdata,
+                                  nullptr, 0);
+        }
+        else
+        {
+            delete context;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
+
+    void parseServerHeaderOptions(OCClientResponse* clientResponse,
+                    HeaderOptions& serverHeaderOptions)
+    {
+        if (clientResponse)
+        {
+            // Parse header options from server
+            uint16_t optionID;
+            std::string optionData;
+
+            for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
+            {
+                optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
+                optionData = reinterpret_cast<const char*>
+                                (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
+                HeaderOption::OCHeaderOption headerOption(optionID, optionData);
+                serverHeaderOptions.push_back(headerOption);
+            }
+        }
+        else
+        {
+            // clientResponse is invalid
+            // TODO check proper logging
+            std::cout << " Invalid response " << std::endl;
+        }
+    }
+
+#ifdef WITH_MQ
+    OCStackApplicationResult createMQTopicCallback(void* ctx, OCDoHandle /*handle*/,
+                    OCClientResponse* clientResponse)
+    {
+        ClientCallbackContext::MQTopicContext* context =
+            static_cast<ClientCallbackContext::MQTopicContext*>(ctx);
+        HeaderOptions serverHeaderOptions;
+
+        if (!clientResponse || !context)
+        {
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        std::string createdUri;
+        bool isLocationOption = false;
+        OCStackResult result = clientResponse->result;
+        if (OC_STACK_OK               == result ||
+            OC_STACK_RESOURCE_CREATED == result)
+        {
+            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+
+            for (auto headerOption : serverHeaderOptions)
+            {
+                if (HeaderOption::LOCATION_PATH_OPTION_ID == headerOption.getOptionID())
+                {
+                    createdUri += "/";
+                    createdUri += headerOption.getOptionData();
+                    if (!isLocationOption)
+                    {
+                        isLocationOption = true;
+                    }
+                }
+            }
+        }
+
+        if (!isLocationOption && NULL != clientResponse->resourceUri)
+        {
+            createdUri = std::string(clientResponse->resourceUri);
+        }
+
+        auto clientWrapper = context->clientWrapper.lock();
+
+        if (!clientWrapper)
+        {
+            oclog() << "createMQTopicCallback(): failed to get a shared_ptr to the client wrapper"
+                    << std::flush;
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        try
+        {
+            if (OC_STACK_OK               == result ||
+                OC_STACK_RESOURCE_CREATED == result)
+            {
+                ListenOCContainer container(clientWrapper, clientResponse->devAddr,
+                                            createdUri);
+                for (auto resource : container.Resources())
+                {
+                    std::thread exec(context->callback, result,
+                                     createdUri,
+                                     resource);
+                    exec.detach();
+                }
+            }
+            else
+            {
+                OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
+                std::thread exec(context->callback, result,
+                                 createdUri,
+                                 nullptr);
+                exec.detach();
+            }
+        }
+        catch (std::exception &e)
         {
-            oclog() <<"Exception in listenDeviceCallback, ignoring response: "
-                <<e.what() <<std::flush;
+            oclog() << "Exception in createMQTopicCallback, ignoring response: "
+                    << e.what() << std::flush;
         }
-
-        return OC_STACK_KEEP_TRANSACTION;
+        return OC_STACK_DELETE_TRANSACTION;
     }
 
-    OCStackResult InProcClientWrapper::ListenForDevice(
-            const std::string& serviceUrl,
-            const std::string& deviceURI,
-            OCConnectivityType connectivityType,
-            FindDeviceCallback& callback,
-            QualityOfService QoS)
+    OCStackResult InProcClientWrapper::PutMQTopicRepresentation(
+                const OCDevAddr& devAddr,
+                const std::string& uri,
+                const OCRepresentation& rep,
+                const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+                MQTopicCallback& callback, QualityOfService QoS)
     {
         if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
         OCStackResult result;
-        ostringstream deviceUri;
-        deviceUri << serviceUrl << deviceURI;
-
-        ClientCallbackContext::DeviceListenContext* context =
-            new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
+        ClientCallbackContext::MQTopicContext* ctx =
+                new ClientCallbackContext::MQTopicContext(callback, shared_from_this());
         OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(ctx),
+        cbdata.cb      = createMQTopicCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::MQTopicContext*)c;};
 
-        cbdata.context = static_cast<void*>(context),
-        cbdata.cb      = listenDeviceCallback;
-        cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
+        std::string url = assembleSetResourceUri(uri, queryParams);
 
         auto cLock = m_csdkLock.lock();
+
         if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
-            result = OCDoResource(nullptr, OC_REST_DISCOVER,
-                                  deviceUri.str().c_str(),
-                                  nullptr, nullptr, connectivityType,
+            OCHeaderOption options[MAX_HEADER_OPTIONS];
+
+            result = OCDoResource(nullptr, OC_REST_PUT,
+                                  url.c_str(), &devAddr,
+                                  assembleSetResourcePayload(rep),
+                                  CT_DEFAULT,
                                   static_cast<OCQualityOfService>(QoS),
                                   &cbdata,
-                                  nullptr, 0);
+                                  assembleHeaderOptions(options, headerOptions),
+                                  headerOptions.size());
         }
         else
         {
-            delete context;
+            delete ctx;
             result = OC_STACK_ERROR;
         }
-        return result;
-    }
-
-    void parseServerHeaderOptions(OCClientResponse* clientResponse,
-                    HeaderOptions& serverHeaderOptions)
-    {
-        if (clientResponse)
-        {
-            // Parse header options from server
-            uint16_t optionID;
-            std::string optionData;
 
-            for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
-            {
-                optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
-                optionData = reinterpret_cast<const char*>
-                                (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
-                HeaderOption::OCHeaderOption headerOption(optionID, optionData);
-                serverHeaderOptions.push_back(headerOption);
-            }
-        }
-        else
-        {
-            // clientResponse is invalid
-            // TODO check proper logging
-            std::cout << " Invalid response " << std::endl;
-        }
+        return result;
     }
-
+#endif
     OCStackApplicationResult getResourceCallback(void* ctx,
                                                  OCDoHandle /*handle*/,
         OCClientResponse* clientResponse)
     {
         ClientCallbackContext::GetContext* context =
             static_cast<ClientCallbackContext::GetContext*>(ctx);
-
         OCRepresentation rep;
         HeaderOptions serverHeaderOptions;
         OCStackResult result = clientResponse->result;
-        if (result == OC_STACK_OK)
+
+        parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+        try
         {
-            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
-            try
-            {
-                rep = parseGetSetCallback(clientResponse);
-            }
-            catch(OC::OCException& e)
-            {
-                result = e.code();
-            }
+            rep = parseGetSetCallback(clientResponse);
+        }
+        catch(OC::OCException& e)
+        {
+            result = e.code();
         }
 
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, serverHeaderOptions, rep, result);
         exec.detach();
         return OC_STACK_DELETE_TRANSACTION;
@@ -447,6 +930,7 @@ namespace OC
         const OCDevAddr& devAddr,
         const std::string& resourceUri,
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+        OCConnectivityType connectivityType,
         GetCallback& callback, QualityOfService QoS)
     {
         if (!callback)
@@ -456,12 +940,12 @@ namespace OC
         OCStackResult result;
         ClientCallbackContext::GetContext* ctx =
             new ClientCallbackContext::GetContext(callback);
+
         OCCallbackData cbdata;
-        cbdata.context = static_cast<void*>(ctx),
+        cbdata.context = static_cast<void*>(ctx);
         cbdata.cb      = getResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 
-
         std::string uri = assembleSetResourceUri(resourceUri, queryParams);
 
         auto cLock = m_csdkLock.lock();
@@ -475,7 +959,7 @@ namespace OC
                                   nullptr, OC_REST_GET,
                                   uri.c_str(),
                                   &devAddr, nullptr,
-                                  CT_DEFAULT,
+                                  connectivityType,
                                   static_cast<OCQualityOfService>(QoS),
                                   &cbdata,
                                   assembleHeaderOptions(options, headerOptions),
@@ -500,22 +984,18 @@ namespace OC
         HeaderOptions serverHeaderOptions;
 
         OCStackResult result = clientResponse->result;
-        if (OC_STACK_OK               == result ||
-            OC_STACK_RESOURCE_CREATED == result ||
-            OC_STACK_RESOURCE_DELETED == result ||
-            OC_STACK_RESOURCE_CHANGED == result)
+
+        parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+        try
         {
-            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
-            try
-            {
-                attrs = parseGetSetCallback(clientResponse);
-            }
-            catch(OC::OCException& e)
-            {
-                result = e.code();
-            }
+            attrs = parseGetSetCallback(clientResponse);
+        }
+        catch(OC::OCException& e)
+        {
+            result = e.code();
         }
 
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, serverHeaderOptions, attrs, result);
         exec.detach();
         return OC_STACK_DELETE_TRANSACTION;
@@ -559,6 +1039,47 @@ namespace OC
         return ret;
     }
 
+    std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
+        const QueryParamsList& queryParams)
+    {
+        if (!uri.empty())
+        {
+            if (uri.back() == '/')
+            {
+                uri.resize(uri.size() - 1);
+            }
+        }
+
+        ostringstream paramsList;
+        if (queryParams.size() > 0)
+        {
+            paramsList << '?';
+        }
+
+        for (auto& param : queryParams)
+        {
+            for (auto& paramList : param.second)
+            {
+                paramsList << param.first << '=' << paramList << ';';
+            }
+        }
+
+        std::string queryString = paramsList.str();
+
+        if (queryString.empty())
+        {
+            return uri;
+        }
+
+        if (queryString.back() == ';')
+        {
+            queryString.resize(queryString.size() - 1);
+        }
+
+        std::string ret = uri + queryString;
+        return ret;
+    }
+
     OCPayload* InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
     {
         MessageContainer ocInfo;
@@ -576,6 +1097,7 @@ namespace OC
         const std::string& uri,
         const OCRepresentation& rep,
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+        OCConnectivityType connectivityType,
         PostCallback& callback, QualityOfService QoS)
     {
         if (!callback)
@@ -602,7 +1124,7 @@ namespace OC
             result = OCDoResource(nullptr, OC_REST_POST,
                                   url.c_str(), &devAddr,
                                   assembleSetResourcePayload(rep),
-                                  CT_DEFAULT,
+                                  connectivityType,
                                   static_cast<OCQualityOfService>(QoS),
                                   &cbdata,
                                   assembleHeaderOptions(options, headerOptions),
@@ -672,10 +1194,9 @@ namespace OC
             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
         HeaderOptions serverHeaderOptions;
 
-        if (clientResponse->result == OC_STACK_OK)
-        {
-            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
-        }
+        parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
         exec.detach();
         return OC_STACK_DELETE_TRANSACTION;
@@ -685,6 +1206,7 @@ namespace OC
         const OCDevAddr& devAddr,
         const std::string& uri,
         const HeaderOptions& headerOptions,
+        OCConnectivityType connectivityType,
         DeleteCallback& callback,
         QualityOfService /*QoS*/)
     {
@@ -712,7 +1234,7 @@ namespace OC
             result = OCDoResource(nullptr, OC_REST_DELETE,
                                   uri.c_str(), &devAddr,
                                   nullptr,
-                                  CT_DEFAULT,
+                                  connectivityType,
                                   static_cast<OCQualityOfService>(m_cfg.QoS),
                                   &cbdata,
                                   assembleHeaderOptions(options, headerOptions),
@@ -737,25 +1259,26 @@ namespace OC
         HeaderOptions serverHeaderOptions;
         uint32_t sequenceNumber = clientResponse->sequenceNumber;
         OCStackResult result = clientResponse->result;
-        if (clientResponse->result == OC_STACK_OK)
+
+        parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+        try
         {
-            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
-            try
-            {
-                attrs = parseGetSetCallback(clientResponse);
-            }
-            catch(OC::OCException& e)
-            {
-                result = e.code();
-            }
+            attrs = parseGetSetCallback(clientResponse);
         }
+        catch(OC::OCException& e)
+        {
+            result = e.code();
+        }
+
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, serverHeaderOptions, attrs,
                     result, sequenceNumber);
         exec.detach();
-        if (sequenceNumber == OC_OBSERVE_DEREGISTER)
+        if (sequenceNumber == MAX_SEQUENCE_NUMBER + 1)
         {
             return OC_STACK_DELETE_TRANSACTION;
         }
+
         return OC_STACK_KEEP_TRANSACTION;
     }
 
@@ -860,6 +1383,7 @@ namespace OC
          */
         std::string url = clientResponse->devAddr.addr;
 
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, clientResponse->result,
                     clientResponse->sequenceNumber, url);
 
@@ -925,6 +1449,53 @@ namespace OC
         return result;
     }
 
+#ifdef WITH_CLOUD
+    OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
+                                                               const std::string& host,
+                                                               const std::vector<std::string>& di,
+                                                               OCConnectivityType connectivityType,
+                                                               ObserveCallback& callback)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+        OCStackResult result;
+
+        ClientCallbackContext::ObserveContext* ctx =
+            new ClientCallbackContext::ObserveContext(callback);
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(ctx),
+        cbdata.cb      = observeResourceCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+
+            std::ostringstream os;
+            os << host << OC_RSRVD_DEVICE_PRESENCE_URI;
+            QueryParamsList queryParams({{OC_RSRVD_DEVICE_ID, di}});
+            std::string url = assembleSetResourceUri(os.str(), queryParams);
+
+            result = OCDoResource(handle, OC_REST_OBSERVE,
+                                  url.c_str(), nullptr,
+                                  nullptr, connectivityType,
+                                  OC_LOW_QOS, &cbdata,
+                                  nullptr, 0);
+        }
+        else
+        {
+            delete ctx;
+            result = OC_STACK_ERROR;
+        }
+
+        return result;
+    }
+#endif
+
     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
     {
         qos = m_cfg.QoS;
@@ -947,7 +1518,9 @@ namespace OC
             options[i].protocolID = OC_COAP_ID;
             options[i].optionID = it->getOptionID();
             options[i].optionLength = it->getOptionData().length() + 1;
-            strcpy((char*)options[i].optionData, (it->getOptionData().c_str()));
+            strncpy((char*)options[i].optionData, it->getOptionData().c_str(),
+                sizeof(options[i].optionLength) -1 );
+            options[i].optionData[sizeof(options[i].optionLength) - 1] = 0;
             i++;
         }
 
@@ -1002,6 +1575,7 @@ namespace OC
                     << std::flush;
             }
             else {
+                OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
                 convert(list, dpDeviceList);
                 std::thread exec(callback, dpDeviceList);
                 exec.detach();
@@ -1037,10 +1611,10 @@ namespace OC
             if (NULL == list)
             {
                 result = OC_STACK_NO_RESOURCE;
-                oclog() << "findDirectPairingDevices(): No device found for direct pairing"
-                    << std::flush;
+                OIC_LOG_V(DEBUG, TAG, "%s: No device found for direct pairing", __func__);
             }
             else {
+                OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
                 convert(list, dpDeviceList);
                 std::thread exec(callback, dpDeviceList);
                 exec.detach();
@@ -1062,6 +1636,7 @@ namespace OC
         ClientCallbackContext::DirectPairingContext* context =
             static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
 
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
         std::thread exec(context->callback, cloneDevice(peer), result);
         exec.detach();
     }
@@ -1093,4 +1668,100 @@ namespace OC
         }
         return result;
     }
+#ifdef TCP_ADAPTER
+    OCStackApplicationResult KeepAliveRespCallback(void* ctx,
+                                                 OCDoHandle /*handle*/,
+        OCClientResponse* clientResponse)
+    {
+        ClientCallbackContext::KeepAliveContext* context =
+            static_cast<ClientCallbackContext::KeepAliveContext*>(ctx);
+        OCRepresentation attrs;
+        OCStackResult result = clientResponse->result;
+
+        try
+        {
+            attrs = parseGetSetCallback(clientResponse);
+        }
+        catch(OC::OCException& e)
+        {
+            result = e.code();
+        }
+
+        OIC_LOG_V(DEBUG, TAG, "%s: call response callback", __func__);
+        std::thread exec(context->callback, result, attrs);
+        exec.detach();
+        return OC_STACK_DELETE_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::findKeepAliveResource(std::string host,
+                                                             KeepAliveCallback resultCallback)
+    {
+        if (host.empty() || !resultCallback)
+        {
+            oclog() << "Invalid parameters" << std::flush;
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result = OC_STACK_ERROR;
+
+        ClientCallbackContext::KeepAliveContext* ctx =
+                       new ClientCallbackContext::KeepAliveContext(resultCallback);
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(ctx),
+        cbdata.cb      = KeepAliveRespCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            result = OCFindKeepAliveResource(nullptr, host.c_str(), &cbdata);
+        }
+        else
+        {
+            delete ctx;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
+
+    OCStackResult InProcClientWrapper::sendKeepAliveRequest(std::string host,
+                                                            const OCRepresentation& rep,
+                                                            KeepAliveCallback resultCallback)
+    {
+        if (!resultCallback)
+        {
+            oclog() << "Invalid parameters" << std::flush;
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result = OC_STACK_ERROR;
+
+        ClientCallbackContext::KeepAliveContext* ctx = new ClientCallbackContext::KeepAliveContext(resultCallback);
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(ctx),
+        cbdata.cb      = KeepAliveRespCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::KeepAliveContext*)c;};
+
+        auto cLock = m_csdkLock.lock();
+
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            OCRepPayload *payload = rep.getPayload();
+            result = OCSendKeepAliveRequest (nullptr, host.c_str(), (OCPayload*)payload, &cbdata);
+            if (result != OC_STACK_OK)
+            {
+                OCRepPayloadDestroy(payload);
+            }
+        }
+        else
+        {
+            delete ctx;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
+#endif
 }