Merge "Merge branch 'master' into extended-easysetup" into extended-easysetup
[platform/upstream/iotivity.git] / resource / src / InProcClientWrapper.cpp
index 3de692b..8a892c6 100644 (file)
@@ -37,7 +37,7 @@ 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
 
-        if(m_cfg.mode == ModeType::Client)
+        if (m_cfg.mode == ModeType::Client)
         {
             OCTransportFlags serverFlags =
                             static_cast<OCTransportFlags>(m_cfg.serverConnectivity & CT_MASK_FLAGS);
@@ -45,7 +45,7 @@ namespace OC
                             static_cast<OCTransportFlags>(m_cfg.clientConnectivity & CT_MASK_FLAGS);
             OCStackResult result = OCInit1(OC_CLIENT, serverFlags, clientFlags);
 
-            if(OC_STACK_OK != result)
+            if (OC_STACK_OK != result)
             {
                 throw InitializeException(OC::InitException::STACK_INIT_ERROR, result);
             }
@@ -57,7 +57,7 @@ namespace OC
 
     InProcClientWrapper::~InProcClientWrapper()
     {
-        if(m_threadRun && m_listeningThread.joinable())
+        if (m_threadRun && m_listeningThread.joinable())
         {
             m_threadRun = false;
             m_listeningThread.join();
@@ -65,7 +65,7 @@ namespace OC
 
         // only stop if we are the ones who actually called 'init'.  We are counting
         // on the server to do the stop.
-        if(m_cfg.mode == ModeType::Client)
+        if (m_cfg.mode == ModeType::Client)
         {
             OCStop();
         }
@@ -77,7 +77,7 @@ namespace OC
         {
             OCStackResult result;
             auto cLock = m_csdkLock.lock();
-            if(cLock)
+            if (cLock)
             {
                 std::lock_guard<std::recursive_mutex> lock(*cLock);
                 result = OCProcess();
@@ -87,7 +87,7 @@ namespace OC
                 result = OC_STACK_ERROR;
             }
 
-            if(result != OC_STACK_OK)
+            if (result != OC_STACK_OK)
             {
                 // TODO: do something with result if failed?
             }
@@ -99,7 +99,7 @@ namespace OC
 
     OCRepresentation parseGetSetCallback(OCClientResponse* clientResponse)
     {
-        if(clientResponse->payload == nullptr ||
+        if (clientResponse->payload == nullptr ||
                 (
                     clientResponse->payload->type != PAYLOAD_TYPE_DEVICE &&
                     clientResponse->payload->type != PAYLOAD_TYPE_PLATFORM &&
@@ -116,7 +116,7 @@ namespace OC
         //OCPayloadDestroy(clientResponse->payload);
 
         std::vector<OCRepresentation>::const_iterator it = oc.representations().begin();
-        if(it == oc.representations().end())
+        if (it == oc.representations().end())
         {
             return OCRepresentation();
         }
@@ -140,7 +140,7 @@ namespace OC
         ClientCallbackContext::ListenContext* context =
             static_cast<ClientCallbackContext::ListenContext*>(ctx);
 
-        if(clientResponse->result != OC_STACK_OK)
+        if (clientResponse->result != OC_STACK_OK)
         {
             oclog() << "listenCallback(): failed to create resource. clientResponse: "
                     << clientResponse->result
@@ -149,7 +149,7 @@ namespace OC
             return OC_STACK_KEEP_TRANSACTION;
         }
 
-        if(!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
+        if (!clientResponse->payload || clientResponse->payload->type != PAYLOAD_TYPE_DISCOVERY)
         {
             oclog() << "listenCallback(): clientResponse payload was null or the wrong type"
                 << std::flush;
@@ -158,7 +158,7 @@ namespace OC
 
         auto clientWrapper = context->clientWrapper.lock();
 
-        if(!clientWrapper)
+        if (!clientWrapper)
         {
             oclog() << "listenCallback(): failed to get a shared_ptr to the client wrapper"
                     << std::flush;
@@ -169,6 +169,7 @@ namespace OC
             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);
@@ -241,7 +242,7 @@ namespace OC
             OCConnectivityType connectivityType,
             FindCallback& callback, QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -253,12 +254,12 @@ namespace OC
         ClientCallbackContext::ListenContext* context =
             new ClientCallbackContext::ListenContext(callback, shared_from_this());
         OCCallbackData cbdata;
-        cbdata.context = (void*)context;
+        cbdata.context = static_cast<void*>(context),
         cbdata.cb      = listenCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext*)c;};
 
         auto cLock = m_csdkLock.lock();
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             result = OCDoResource(nullptr, OC_REST_DISCOVER,
@@ -324,6 +325,101 @@ namespace OC
         }
         return result;
     }
+#ifdef WITH_MQ
+    OCStackApplicationResult listenMQCallback(void* ctx, OCDoHandle /*handle*/,
+            OCClientResponse* clientResponse)
+    {
+        ClientCallbackContext::ListenContext* context =
+            static_cast<ClientCallbackContext::ListenContext*>(ctx);
+
+        if (!clientResponse)
+        {
+            return OC_STACK_KEEP_TRANSACTION;
+        }
+
+        if (clientResponse->result != OC_STACK_OK)
+        {
+            oclog() << "listenMQCallback(): failed to create resource. clientResponse: "
+                    << clientResponse->result
+                    << std::flush;
+
+            return OC_STACK_KEEP_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_KEEP_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, resource);
+                exec.detach();
+            }
+        }
+        catch (std::exception &e){
+            oclog() << "Exception in listCallback, ignoring response: "
+                    << e.what() << std::flush;
+        }
+
+
+        return OC_STACK_KEEP_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::ListenForMQTopic(
+            const OCDevAddr& devAddr,
+            const std::string& resourceUri,
+            const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+            FindCallback& callback, QualityOfService QoS)
+    {
+        oclog() << "ListenForMQTopic()" << std::flush;
+
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        ClientCallbackContext::ListenContext* context =
+            new ClientCallbackContext::ListenContext(callback, shared_from_this());
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(context),
+        cbdata.cb      = listenMQCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::ListenContext*)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*/,
@@ -354,7 +450,7 @@ namespace OC
             FindDeviceCallback& callback,
             QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -366,12 +462,12 @@ namespace OC
             new ClientCallbackContext::DeviceListenContext(callback, shared_from_this());
         OCCallbackData cbdata;
 
-        cbdata.context = (void*)context;
+        cbdata.context = static_cast<void*>(context),
         cbdata.cb      = listenDeviceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeviceListenContext*)c;};
 
         auto cLock = m_csdkLock.lock();
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             result = OCDoResource(nullptr, OC_REST_DISCOVER,
@@ -392,7 +488,7 @@ namespace OC
     void parseServerHeaderOptions(OCClientResponse* clientResponse,
                     HeaderOptions& serverHeaderOptions)
     {
-        if(clientResponse)
+        if (clientResponse)
         {
             // Parse header options from server
             uint16_t optionID;
@@ -415,6 +511,135 @@ namespace OC
         }
     }
 
+#ifdef WITH_MQ
+    OCStackApplicationResult createMQTopicCallback(void* ctx, OCDoHandle /*handle*/,
+                    OCClientResponse* clientResponse)
+    {
+        ClientCallbackContext::CreateMQTopicContext* context =
+            static_cast<ClientCallbackContext::CreateMQTopicContext*>(ctx);
+        OCRepresentation rep;
+        HeaderOptions serverHeaderOptions;
+
+        if (!clientResponse)
+        {
+            return OC_STACK_DELETE_TRANSACTION;
+        }
+
+        std::string createdUri;
+        OCStackResult result = clientResponse->result;
+        if (OC_STACK_OK               == result ||
+            OC_STACK_RESOURCE_CREATED == result)
+        {
+            parseServerHeaderOptions(clientResponse, serverHeaderOptions);
+            try
+            {
+                rep = parseGetSetCallback(clientResponse);
+            }
+            catch(OC::OCException& e)
+            {
+                result = e.code();
+            }
+
+            bool isLocationOption = false;
+            for (auto headerOption : serverHeaderOptions)
+            {
+                if (HeaderOption::LOCATION_PATH_OPTION_ID == headerOption.getOptionID())
+                {
+                    createdUri += "/";
+                    createdUri += headerOption.getOptionData();
+                    if (!isLocationOption)
+                    {
+                        isLocationOption = true;
+                    }
+                }
+            }
+
+            if (!isLocationOption)
+            {
+                createdUri = 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, serverHeaderOptions, rep, result, resource);
+                    exec.detach();
+                }
+            }
+            else
+            {
+                std::thread exec(context->callback, serverHeaderOptions, rep, result, nullptr);
+                exec.detach();
+            }
+        }
+        catch (std::exception &e){
+            oclog() << "Exception in createMQTopicCallback, ignoring response: "
+                    << e.what() << std::flush;
+        }
+        return OC_STACK_DELETE_TRANSACTION;
+    }
+
+    OCStackResult InProcClientWrapper::PutMQTopicRepresentation(
+                const OCDevAddr& devAddr,
+                const std::string& uri,
+                const OCRepresentation& rep,
+                const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+                MQCreateTopicCallback& callback, QualityOfService QoS)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+        OCStackResult result;
+        ClientCallbackContext::CreateMQTopicContext* ctx =
+                new ClientCallbackContext::CreateMQTopicContext(callback, shared_from_this());
+        OCCallbackData cbdata;
+        cbdata.context = static_cast<void*>(ctx),
+        cbdata.cb      = createMQTopicCallback;
+        cbdata.cd      = [](void* c){delete (ClientCallbackContext::CreateMQTopicContext*)c;};
+
+        std::string url = assembleSetResourceUri(uri, queryParams);
+
+        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_PUT,
+                                  url.c_str(), &devAddr,
+                                  assembleSetResourcePayload(rep),
+                                  CT_DEFAULT,
+                                  static_cast<OCQualityOfService>(QoS),
+                                  &cbdata,
+                                  assembleHeaderOptions(options, headerOptions),
+                                  headerOptions.size());
+        }
+        else
+        {
+            delete ctx;
+            result = OC_STACK_ERROR;
+        }
+
+        return result;
+    }
+#endif
     OCStackApplicationResult getResourceCallback(void* ctx,
                                                  OCDoHandle /*handle*/,
         OCClientResponse* clientResponse)
@@ -425,7 +650,7 @@ namespace OC
         OCRepresentation rep;
         HeaderOptions serverHeaderOptions;
         OCStackResult result = clientResponse->result;
-        if(result == OC_STACK_OK)
+        if (result == OC_STACK_OK)
         {
             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
             try
@@ -449,7 +674,7 @@ namespace OC
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
         GetCallback& callback, QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -457,7 +682,7 @@ namespace OC
         ClientCallbackContext::GetContext* ctx =
             new ClientCallbackContext::GetContext(callback);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = getResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::GetContext*)c;};
 
@@ -466,7 +691,7 @@ namespace OC
 
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             OCHeaderOption options[MAX_HEADER_OPTIONS];
@@ -502,7 +727,8 @@ namespace OC
         OCStackResult result = clientResponse->result;
         if (OC_STACK_OK               == result ||
             OC_STACK_RESOURCE_CREATED == result ||
-            OC_STACK_RESOURCE_DELETED == result)
+            OC_STACK_RESOURCE_DELETED == result ||
+            OC_STACK_RESOURCE_CHANGED == result)
         {
             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
             try
@@ -523,33 +749,79 @@ namespace OC
     std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
         const QueryParamsMap& queryParams)
     {
-        if(!uri.empty())
+        if (!uri.empty())
         {
-            if(uri.back() == '/')
+            if (uri.back() == '/')
             {
-                uri.resize(uri.size()-1);
+                uri.resize(uri.size() - 1);
             }
         }
 
-        std::ostringstream paramsList;
-        if(queryParams.size() > 0)
+        ostringstream paramsList;
+        if (queryParams.size() > 0)
         {
             paramsList << '?';
         }
 
-        for(auto& param : queryParams)
+        for (auto& param : queryParams)
         {
             paramsList << param.first <<'='<<param.second<<';';
         }
 
         std::string queryString = paramsList.str();
 
-        if(queryString.empty())
+        if (queryString.empty())
+        {
+            return uri;
+        }
+
+        if (queryString.back() == ';')
+        {
+            queryString.resize(queryString.size() - 1);
+        }
+
+        std::string ret = uri + queryString;
+        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;
+                if (paramList != param.second.back())
+                {
+                    paramsList << '&';
+                }
+            }
+            paramsList << ';';
+        }
+
+        std::string queryString = paramsList.str();
+
+        if (queryString.empty())
         {
             return uri;
         }
 
-        if(queryString.back() == ';')
+        if (queryString.back() == ';')
         {
             queryString.resize(queryString.size() - 1);
         }
@@ -575,16 +847,17 @@ namespace OC
         const std::string& uri,
         const OCRepresentation& rep,
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
+        OCConnectivityType connectivityType,
         PostCallback& callback, QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
         OCStackResult result;
         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = setResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 
@@ -593,7 +866,7 @@ namespace OC
 
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             OCHeaderOption options[MAX_HEADER_OPTIONS];
@@ -601,7 +874,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),
@@ -623,14 +896,14 @@ namespace OC
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
         PutCallback& callback, QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
         OCStackResult result;
         ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext(callback);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = setResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SetContext*)c;};
 
@@ -639,7 +912,7 @@ namespace OC
 
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             OCDoHandle handle;
@@ -671,7 +944,7 @@ namespace OC
             static_cast<ClientCallbackContext::DeleteContext*>(ctx);
         HeaderOptions serverHeaderOptions;
 
-        if(clientResponse->result == OC_STACK_OK)
+        if (clientResponse->result == OC_STACK_OK)
         {
             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
         }
@@ -687,7 +960,7 @@ namespace OC
         DeleteCallback& callback,
         QualityOfService /*QoS*/)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -695,14 +968,14 @@ namespace OC
         ClientCallbackContext::DeleteContext* ctx =
             new ClientCallbackContext::DeleteContext(callback);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = deleteResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::DeleteContext*)c;};
 
 
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             OCHeaderOption options[MAX_HEADER_OPTIONS];
 
@@ -736,7 +1009,7 @@ namespace OC
         HeaderOptions serverHeaderOptions;
         uint32_t sequenceNumber = clientResponse->sequenceNumber;
         OCStackResult result = clientResponse->result;
-        if(clientResponse->result == OC_STACK_OK)
+        if (clientResponse->result == OC_STACK_OK)
         {
             parseServerHeaderOptions(clientResponse, serverHeaderOptions);
             try
@@ -751,7 +1024,7 @@ namespace OC
         std::thread exec(context->callback, serverHeaderOptions, attrs,
                     result, sequenceNumber);
         exec.detach();
-        if(sequenceNumber == OC_OBSERVE_DEREGISTER)
+        if (sequenceNumber == OC_OBSERVE_DEREGISTER)
         {
             return OC_STACK_DELETE_TRANSACTION;
         }
@@ -764,7 +1037,7 @@ namespace OC
         const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
         ObserveCallback& callback, QualityOfService QoS)
     {
-        if(!callback)
+        if (!callback)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -773,7 +1046,7 @@ namespace OC
         ClientCallbackContext::ObserveContext* ctx =
             new ClientCallbackContext::ObserveContext(callback);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = observeResourceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::ObserveContext*)c;};
 
@@ -796,7 +1069,7 @@ namespace OC
 
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             OCHeaderOption options[MAX_HEADER_OPTIONS];
@@ -829,7 +1102,7 @@ namespace OC
         OCStackResult result;
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             OCHeaderOption options[MAX_HEADER_OPTIONS];
@@ -871,7 +1144,7 @@ namespace OC
         const std::string& host, const std::string& resourceType,
         OCConnectivityType connectivityType, SubscribeCallback& presenceHandler)
     {
-        if(!presenceHandler)
+        if (!presenceHandler)
         {
             return OC_STACK_INVALID_PARAM;
         }
@@ -879,7 +1152,7 @@ namespace OC
         ClientCallbackContext::SubscribePresenceContext* ctx =
             new ClientCallbackContext::SubscribePresenceContext(presenceHandler);
         OCCallbackData cbdata;
-        cbdata.context = (void*)ctx;
+        cbdata.context = static_cast<void*>(ctx),
         cbdata.cb      = subscribePresenceCallback;
         cbdata.cd      = [](void* c){delete (ClientCallbackContext::SubscribePresenceContext*)c;};
 
@@ -889,12 +1162,12 @@ namespace OC
         std::ostringstream os;
         os << host << OC_RSRVD_PRESENCE_URI;
 
-        if(!resourceType.empty())
+        if (!resourceType.empty())
         {
             os << "?rt=" << resourceType;
         }
 
-        if(!cLock)
+        if (!cLock)
         {
             delete ctx;
             return OC_STACK_ERROR;
@@ -911,7 +1184,7 @@ namespace OC
         OCStackResult result;
         auto cLock = m_csdkLock.lock();
 
-        if(cLock)
+        if (cLock)
         {
             std::lock_guard<std::recursive_mutex> lock(*cLock);
             result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
@@ -924,6 +1197,50 @@ namespace OC
         return result;
     }
 
+    OCStackResult InProcClientWrapper::SubscribeDevicePresence(OCDoHandle* handle,
+                                                               const std::string& host,
+                                                               const QueryParamsList& queryParams,
+                                                               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 << OCF_RSRVD_DEVICE_PRESENCE_URI;
+            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;
+    }
+
     OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
     {
         qos = m_cfg.QoS;
@@ -935,7 +1252,7 @@ namespace OC
     {
         int i = 0;
 
-        if( headerOptions.size() == 0)
+        if ( headerOptions.size() == 0)
         {
             return nullptr;
         }
@@ -952,4 +1269,144 @@ namespace OC
 
         return options;
     }
+
+    std::shared_ptr<OCDirectPairing> cloneDevice(const OCDPDev_t* dev)
+    {
+        if (!dev)
+        {
+            return nullptr;
+        }
+
+        OCDPDev_t* result = new OCDPDev_t(*dev);
+        result->prm = new OCPrm_t[dev->prmLen];
+        memcpy(result->prm, dev->prm, sizeof(OCPrm_t)*dev->prmLen);
+        return std::shared_ptr<OCDirectPairing>(new OCDirectPairing(result));
+    }
+
+    void InProcClientWrapper::convert(const OCDPDev_t *list, PairedDevices& dpList)
+    {
+        while(list)
+        {
+            dpList.push_back(cloneDevice(list));
+            list = list->next;
+        }
+    }
+
+    OCStackResult InProcClientWrapper::FindDirectPairingDevices(unsigned short waittime,
+            GetDirectPairedCallback& callback)
+    {
+        if (!callback || 0 == waittime)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result = OC_STACK_ERROR;
+        const OCDPDev_t *list = nullptr;
+        PairedDevices dpDeviceList;
+
+        auto cLock = m_csdkLock.lock();
+
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+
+            list = OCDiscoverDirectPairingDevices(waittime);
+            if (NULL == list)
+            {
+                result = OC_STACK_NO_RESOURCE;
+                oclog() << "findDirectPairingDevices(): No device found for direct pairing"
+                    << std::flush;
+            }
+            else {
+                convert(list, dpDeviceList);
+                std::thread exec(callback, dpDeviceList);
+                exec.detach();
+                result = OC_STACK_OK;
+            }
+        }
+        else
+        {
+            result = OC_STACK_ERROR;
+        }
+
+        return result;
+    }
+
+    OCStackResult InProcClientWrapper::GetDirectPairedDevices(GetDirectPairedCallback& callback)
+    {
+        if (!callback)
+        {
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result = OC_STACK_ERROR;
+        const OCDPDev_t *list = nullptr;
+        PairedDevices dpDeviceList;
+
+        auto cLock = m_csdkLock.lock();
+
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+
+            list = OCGetDirectPairedDevices();
+            if (NULL == list)
+            {
+                result = OC_STACK_NO_RESOURCE;
+                oclog() << "findDirectPairingDevices(): No device found for direct pairing"
+                    << std::flush;
+            }
+            else {
+                convert(list, dpDeviceList);
+                std::thread exec(callback, dpDeviceList);
+                exec.detach();
+                result = OC_STACK_OK;
+            }
+        }
+        else
+        {
+            result = OC_STACK_ERROR;
+        }
+
+        return result;
+    }
+
+    void directPairingCallback(void *ctx, OCDPDev_t *peer,
+            OCStackResult result)
+    {
+
+        ClientCallbackContext::DirectPairingContext* context =
+            static_cast<ClientCallbackContext::DirectPairingContext*>(ctx);
+
+        std::thread exec(context->callback, cloneDevice(peer), result);
+        exec.detach();
+    }
+
+    OCStackResult InProcClientWrapper::DoDirectPairing(std::shared_ptr<OCDirectPairing> peer,
+            const OCPrm_t& pmSel, const std::string& pinNumber, DirectPairingCallback& callback)
+    {
+        if (!peer || !callback)
+        {
+            oclog() << "Invalid parameters" << std::flush;
+            return OC_STACK_INVALID_PARAM;
+        }
+
+        OCStackResult result = OC_STACK_ERROR;
+        ClientCallbackContext::DirectPairingContext* context =
+            new ClientCallbackContext::DirectPairingContext(callback);
+
+        auto cLock = m_csdkLock.lock();
+        if (cLock)
+        {
+            std::lock_guard<std::recursive_mutex> lock(*cLock);
+            result = OCDoDirectPairing(static_cast<void*>(context), peer->getDev(),
+                    pmSel, const_cast<char*>(pinNumber.c_str()), directPairingCallback);
+        }
+        else
+        {
+            delete context;
+            result = OC_STACK_ERROR;
+        }
+        return result;
+    }
 }