1 //******************************************************************
3 // Copyright 2014 Intel Mobile Communications GmbH All Rights Reserved.
5 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
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
11 // http://www.apache.org/licenses/LICENSE-2.0
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.
19 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
21 #include "InProcClientWrapper.h"
24 #include "OCPlatform.h"
25 #include "OCResource.h"
26 #include <OCSerialization.h>
31 InProcClientWrapper::InProcClientWrapper(
32 std::weak_ptr<std::recursive_mutex> csdkLock, PlatformConfig cfg)
33 : m_threadRun(false), m_csdkLock(csdkLock),
36 // if the config type is server, we ought to never get called. If the config type
37 // is both, we count on the server to run the thread and do the initialize
39 if(m_cfg.mode == ModeType::Client)
41 OCStackResult result = OCInit(m_cfg.ipAddress.c_str(), m_cfg.port, OC_CLIENT);
43 if(OC_STACK_OK != result)
45 throw InitializeException(OC::InitException::STACK_INIT_ERROR, result);
49 m_listeningThread = std::thread(&InProcClientWrapper::listeningFunc, this);
53 InProcClientWrapper::~InProcClientWrapper()
55 if(m_threadRun && m_listeningThread.joinable())
58 m_listeningThread.join();
64 void InProcClientWrapper::listeningFunc()
69 auto cLock = m_csdkLock.lock();
72 std::lock_guard<std::recursive_mutex> lock(*cLock);
77 result = OC_STACK_ERROR;
80 if(result != OC_STACK_OK)
82 // TODO: do something with result if failed?
85 // To minimize CPU utilization we may wish to do this with sleep
86 std::this_thread::sleep_for(std::chrono::milliseconds(10));
90 OCStackApplicationResult listenCallback(void* ctx, OCDoHandle handle,
91 OCClientResponse* clientResponse)
93 ClientCallbackContext::ListenContext* context =
94 static_cast<ClientCallbackContext::ListenContext*>(ctx);
96 if(clientResponse->result != OC_STACK_OK)
98 oclog() << "listenCallback(): failed to create resource. clientResponse: "
99 << clientResponse->result
102 return OC_STACK_KEEP_TRANSACTION;
105 std::stringstream requestStream;
106 requestStream << clientResponse->resJSONPayload;
110 ListenOCContainer container(context->clientWrapper, *clientResponse->addr,
113 // loop to ensure valid construction of all resources
114 for(auto resource : container.Resources())
116 std::thread exec(context->callback, resource);
121 catch(const std::exception& e)
123 oclog() << "listenCallback failed to parse a malformed message: "
125 << std::endl <<std::endl
126 << clientResponse->result
128 return OC_STACK_KEEP_TRANSACTION;
131 return OC_STACK_KEEP_TRANSACTION;
135 OCStackResult InProcClientWrapper::ListenForResource(const std::string& serviceUrl,
136 const std::string& resourceType, FindCallback& callback, QualityOfService QoS)
138 OCStackResult result;
140 OCCallbackData cbdata = {0};
142 ClientCallbackContext::ListenContext* context = new ClientCallbackContext::ListenContext();
143 context->callback = callback;
144 context->clientWrapper = shared_from_this();
146 cbdata.context = static_cast<void*>(context);
147 cbdata.cb = listenCallback;
148 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::ListenContext*>(c);};
150 auto cLock = m_csdkLock.lock();
153 std::lock_guard<std::recursive_mutex> lock(*cLock);
155 result = OCDoResource(&handle, OC_REST_GET,
156 resourceType.c_str(),
158 static_cast<OCQualityOfService>(QoS),
165 result = OC_STACK_ERROR;
170 OCRepresentation parseGetSetCallback(OCClientResponse* clientResponse)
172 if(clientResponse->resJSONPayload == nullptr || clientResponse->resJSONPayload[0] == '\0')
174 return OCRepresentation();
178 oc.setJSONRepresentation(clientResponse->resJSONPayload);
180 std::vector<OCRepresentation>::const_iterator it = oc.representations().begin();
181 if(it == oc.representations().end())
183 return OCRepresentation();
186 // first one is considered the root, everything else is considered a child of this one.
187 OCRepresentation root = *it;
190 std::for_each(it, oc.representations().end(),
191 [&root](const OCRepresentation& repItr)
192 {root.addChild(repItr);});
197 void parseServerHeaderOptions(OCClientResponse* clientResponse,
198 HeaderOptions& serverHeaderOptions)
202 // Parse header options from server
204 std::string optionData;
206 for(int i = 0; i < clientResponse->numRcvdVendorSpecificHeaderOptions; i++)
208 optionID = clientResponse->rcvdVendorSpecificHeaderOptions[i].optionID;
209 optionData = reinterpret_cast<const char*>
210 (clientResponse->rcvdVendorSpecificHeaderOptions[i].optionData);
211 HeaderOption::OCHeaderOption headerOption(optionID, optionData);
212 serverHeaderOptions.push_back(headerOption);
217 // clientResponse is invalid
218 // TODO check proper logging
219 std::cout << " Invalid response " << std::endl;
223 OCStackApplicationResult getResourceCallback(void* ctx, OCDoHandle handle,
224 OCClientResponse* clientResponse)
226 ClientCallbackContext::GetContext* context =
227 static_cast<ClientCallbackContext::GetContext*>(ctx);
229 OCRepresentation rep;
230 HeaderOptions serverHeaderOptions;
231 if(clientResponse->result == OC_STACK_OK)
233 parseServerHeaderOptions(clientResponse, serverHeaderOptions);
234 rep = parseGetSetCallback(clientResponse);
237 std::thread exec(context->callback, serverHeaderOptions, rep, clientResponse->result);
239 return OC_STACK_DELETE_TRANSACTION;
242 OCStackResult InProcClientWrapper::GetResourceRepresentation(const std::string& host,
243 const std::string& uri, const QueryParamsMap& queryParams,
244 const HeaderOptions& headerOptions, GetCallback& callback,
245 QualityOfService QoS)
247 OCStackResult result;
248 OCCallbackData cbdata = {0};
250 ClientCallbackContext::GetContext* ctx = new ClientCallbackContext::GetContext();
251 ctx->callback = callback;
252 cbdata.context = static_cast<void*>(ctx);
253 cbdata.cb = &getResourceCallback;
254 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::GetContext*>(c);};
256 auto cLock = m_csdkLock.lock();
260 std::ostringstream os;
261 os << host << assembleSetResourceUri(uri, queryParams).c_str();
263 std::lock_guard<std::recursive_mutex> lock(*cLock);
265 OCHeaderOption options[MAX_HEADER_OPTIONS];
267 assembleHeaderOptions(options, headerOptions);
268 result = OCDoResource(&handle, OC_REST_GET, os.str().c_str(),
270 static_cast<OCQualityOfService>(QoS),
272 options, headerOptions.size());
277 result = OC_STACK_ERROR;
283 OCStackApplicationResult setResourceCallback(void* ctx, OCDoHandle handle,
284 OCClientResponse* clientResponse)
286 ClientCallbackContext::SetContext* context =
287 static_cast<ClientCallbackContext::SetContext*>(ctx);
288 OCRepresentation attrs;
289 HeaderOptions serverHeaderOptions;
291 if (OC_STACK_OK == clientResponse->result ||
292 OC_STACK_RESOURCE_CREATED == clientResponse->result ||
293 OC_STACK_RESOURCE_DELETED == clientResponse->result)
295 parseServerHeaderOptions(clientResponse, serverHeaderOptions);
296 attrs = parseGetSetCallback(clientResponse);
299 std::thread exec(context->callback, serverHeaderOptions, attrs, clientResponse->result);
301 return OC_STACK_DELETE_TRANSACTION;
304 std::string InProcClientWrapper::assembleSetResourceUri(std::string uri,
305 const QueryParamsMap& queryParams)
307 if(uri.back() == '/')
309 uri.resize(uri.size()-1);
312 ostringstream paramsList;
313 if(queryParams.size() > 0)
318 for(auto& param : queryParams)
320 paramsList << param.first <<'='<<param.second<<'&';
323 std::string queryString = paramsList.str();
324 if(queryString.back() == '&')
326 queryString.resize(queryString.size() - 1);
329 std::string ret = uri + queryString;
333 std::string InProcClientWrapper::assembleSetResourcePayload(const OCRepresentation& rep)
335 MessageContainer ocInfo;
336 ocInfo.addRepresentation(rep);
337 return ocInfo.getJSONRepresentation(OCInfoFormat::IncludeOC);
340 OCStackResult InProcClientWrapper::PostResourceRepresentation(const std::string& host,
341 const std::string& uri, const OCRepresentation& rep,
342 const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
343 PostCallback& callback, QualityOfService QoS)
345 OCStackResult result;
346 OCCallbackData cbdata = {0};
348 ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext();
349 ctx->callback = callback;
350 cbdata.cb = &setResourceCallback;
351 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);};
352 cbdata.context = static_cast<void*>(ctx);
354 // TODO: in the future the cstack should be combining these two strings!
356 os << host << assembleSetResourceUri(uri, queryParams).c_str();
357 // TODO: end of above
359 auto cLock = m_csdkLock.lock();
363 std::lock_guard<std::recursive_mutex> lock(*cLock);
364 OCHeaderOption options[MAX_HEADER_OPTIONS];
367 assembleHeaderOptions(options, headerOptions);
368 result = OCDoResource(&handle, OC_REST_POST,
369 os.str().c_str(), nullptr,
370 assembleSetResourcePayload(rep).c_str(),
371 static_cast<OCQualityOfService>(QoS),
372 &cbdata, options, headerOptions.size());
377 result = OC_STACK_ERROR;
384 OCStackResult InProcClientWrapper::PutResourceRepresentation(const std::string& host,
385 const std::string& uri, const OCRepresentation& rep,
386 const QueryParamsMap& queryParams, const HeaderOptions& headerOptions,
387 PutCallback& callback, QualityOfService QoS)
389 OCStackResult result;
390 OCCallbackData cbdata = {0};
392 ClientCallbackContext::SetContext* ctx = new ClientCallbackContext::SetContext();
393 ctx->callback = callback;
394 cbdata.cb = &setResourceCallback;
395 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::SetContext*>(c);};
396 cbdata.context = static_cast<void*>(ctx);
398 // TODO: in the future the cstack should be combining these two strings!
400 os << host << assembleSetResourceUri(uri, queryParams).c_str();
401 // TODO: end of above
403 auto cLock = m_csdkLock.lock();
407 std::lock_guard<std::recursive_mutex> lock(*cLock);
409 OCHeaderOption options[MAX_HEADER_OPTIONS];
411 assembleHeaderOptions(options, headerOptions);
412 result = OCDoResource(&handle, OC_REST_PUT,
413 os.str().c_str(), nullptr,
414 assembleSetResourcePayload(rep).c_str(),
415 static_cast<OCQualityOfService>(QoS),
417 options, headerOptions.size());
422 result = OC_STACK_ERROR;
428 OCStackApplicationResult deleteResourceCallback(void* ctx, OCDoHandle handle,
429 OCClientResponse* clientResponse)
431 ClientCallbackContext::DeleteContext* context =
432 static_cast<ClientCallbackContext::DeleteContext*>(ctx);
433 HeaderOptions serverHeaderOptions;
435 if(clientResponse->result == OC_STACK_OK)
437 parseServerHeaderOptions(clientResponse, serverHeaderOptions);
439 std::thread exec(context->callback, serverHeaderOptions, clientResponse->result);
441 return OC_STACK_DELETE_TRANSACTION;
444 OCStackResult InProcClientWrapper::DeleteResource(const std::string& host,
445 const std::string& uri, const HeaderOptions& headerOptions,
446 DeleteCallback& callback, QualityOfService QoS)
448 OCStackResult result;
449 OCCallbackData cbdata = {0};
451 ClientCallbackContext::DeleteContext* ctx = new ClientCallbackContext::DeleteContext();
452 ctx->callback = callback;
453 cbdata.cb = &deleteResourceCallback;
454 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::DeleteContext*>(c);};
455 cbdata.context = static_cast<void*>(ctx);
460 auto cLock = m_csdkLock.lock();
464 OCHeaderOption options[MAX_HEADER_OPTIONS];
467 assembleHeaderOptions(options, headerOptions);
469 std::lock_guard<std::recursive_mutex> lock(*cLock);
471 result = OCDoResource(&handle, OC_REST_DELETE,
472 os.str().c_str(), nullptr,
473 nullptr, static_cast<OCQualityOfService>(m_cfg.QoS),
474 &cbdata, options, headerOptions.size());
479 result = OC_STACK_ERROR;
485 OCStackApplicationResult observeResourceCallback(void* ctx, OCDoHandle handle,
486 OCClientResponse* clientResponse)
488 ClientCallbackContext::ObserveContext* context =
489 static_cast<ClientCallbackContext::ObserveContext*>(ctx);
490 OCRepresentation attrs;
491 HeaderOptions serverHeaderOptions;
492 uint32_t sequenceNumber = clientResponse->sequenceNumber;
494 if(clientResponse->result == OC_STACK_OK)
496 parseServerHeaderOptions(clientResponse, serverHeaderOptions);
497 attrs = parseGetSetCallback(clientResponse);
499 std::thread exec(context->callback, serverHeaderOptions, attrs,
500 clientResponse->result, sequenceNumber);
502 return OC_STACK_KEEP_TRANSACTION;
505 OCStackResult InProcClientWrapper::ObserveResource(ObserveType observeType, OCDoHandle* handle,
506 const std::string& host, const std::string& uri, const QueryParamsMap& queryParams,
507 const HeaderOptions& headerOptions, ObserveCallback& callback, QualityOfService QoS)
509 OCStackResult result;
510 OCCallbackData cbdata = {0};
512 ClientCallbackContext::ObserveContext* ctx = new ClientCallbackContext::ObserveContext();
513 ctx->callback = callback;
514 cbdata.context = static_cast<void*>(ctx);
515 cbdata.cb = &observeResourceCallback;
516 cbdata.cd = [](void* c){delete static_cast<ClientCallbackContext::ObserveContext*>(c);};
519 if (observeType == ObserveType::Observe)
521 method = OC_REST_OBSERVE;
523 else if (observeType == ObserveType::ObserveAll)
525 method = OC_REST_OBSERVE_ALL;
529 method = OC_REST_OBSERVE_ALL;
532 auto cLock = m_csdkLock.lock();
536 std::ostringstream os;
537 os << host << assembleSetResourceUri(uri, queryParams).c_str();
539 std::lock_guard<std::recursive_mutex> lock(*cLock);
540 OCHeaderOption options[MAX_HEADER_OPTIONS];
542 assembleHeaderOptions(options, headerOptions);
543 result = OCDoResource(handle, method,
544 os.str().c_str(), nullptr,
546 static_cast<OCQualityOfService>(QoS),
548 options, headerOptions.size());
553 return OC_STACK_ERROR;
559 OCStackResult InProcClientWrapper::CancelObserveResource(OCDoHandle handle,
560 const std::string& host, const std::string& uri, const HeaderOptions& headerOptions,
561 QualityOfService QoS)
563 OCStackResult result;
564 auto cLock = m_csdkLock.lock();
568 std::lock_guard<std::recursive_mutex> lock(*cLock);
569 OCHeaderOption options[MAX_HEADER_OPTIONS];
571 assembleHeaderOptions(options, headerOptions);
572 result = OCCancel(handle, static_cast<OCQualityOfService>(QoS), options,
573 headerOptions.size());
577 result = OC_STACK_ERROR;
583 OCStackApplicationResult subscribePresenceCallback(void* ctx, OCDoHandle handle,
584 OCClientResponse* clientResponse)
586 char stringAddress[DEV_ADDR_SIZE_MAX];
590 if(OCDevAddrToString(clientResponse->addr, stringAddress) == 0 &&
591 OCDevAddrToPort(clientResponse->addr, &port) == 0)
593 os<<stringAddress<<":"<<port;
595 ClientCallbackContext::SubscribePresenceContext* context =
596 static_cast<ClientCallbackContext::SubscribePresenceContext*>(ctx);
598 std::thread exec(context->callback, clientResponse->result,
599 clientResponse->sequenceNumber, os.str());
605 oclog() << "subscribePresenceCallback(): OCDevAddrToString() or OCDevAddrToPort() "
606 <<"failed"<< std::flush;
608 return OC_STACK_KEEP_TRANSACTION;
611 OCStackResult InProcClientWrapper::SubscribePresence(OCDoHandle* handle,
612 const std::string& host, const std::string& resourceType,
613 SubscribeCallback& presenceHandler)
615 OCCallbackData cbdata = {0};
617 ClientCallbackContext::SubscribePresenceContext* ctx =
618 new ClientCallbackContext::SubscribePresenceContext();
619 ctx->callback = presenceHandler;
620 cbdata.cb = &subscribePresenceCallback;
621 cbdata.context = static_cast<void*>(ctx);
622 cbdata.cd = [](void* c)
623 {delete static_cast<ClientCallbackContext::SubscribePresenceContext*>(c);};
624 auto cLock = m_csdkLock.lock();
626 std::ostringstream os;
627 os << host << "/oc/presence";
629 if(!resourceType.empty())
631 os << "?rt=" << resourceType;
637 return OC_STACK_ERROR;
640 return OCDoResource(handle, OC_REST_PRESENCE, os.str().c_str(), nullptr, nullptr,
641 OC_LOW_QOS, &cbdata, NULL, 0);
644 OCStackResult InProcClientWrapper::UnsubscribePresence(OCDoHandle handle)
646 OCStackResult result;
647 auto cLock = m_csdkLock.lock();
651 std::lock_guard<std::recursive_mutex> lock(*cLock);
652 result = OCCancel(handle, OC_LOW_QOS, NULL, 0);
656 result = OC_STACK_ERROR;
662 OCStackResult InProcClientWrapper::GetDefaultQos(QualityOfService& qos)
668 void InProcClientWrapper::assembleHeaderOptions(OCHeaderOption options[],
669 const HeaderOptions& headerOptions)
673 for (auto it=headerOptions.begin(); it != headerOptions.end(); ++it)
675 options[i].protocolID = OC_COAP_ID;
676 options[i].optionID = static_cast<uint16_t>(it->getOptionID());
677 options[i].optionLength = (it->getOptionData()).length() + 1;
678 memcpy(options[i].optionData, (it->getOptionData()).c_str(),
679 (it->getOptionData()).length() + 1);