bridging/include/ConcurrentIotivityUtils.h

   1 //******************************************************************
   2 //
   3 // Copyright 2017 Intel Mobile Communications GmbH All Rights Reserved.
   4 //
   5 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
   6 //
   7 // Licensed under the Apache License, Version 2.0 (the "License");
   8 // you may not use this file except in compliance with the License.
   9 // You may obtain a copy of the License at
  10 //
  11 //      http://www.apache.org/licenses/LICENSE-2.0
  12 //
  13 // Unless required by applicable law or agreed to in writing, software
  14 // distributed under the License is distributed on an "AS IS" BASIS,
  15 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 // See the License for the specific language governing permissions and
  17 // limitations under the License.
  18 //
  19 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
  20 //
  21
  22
  23 #ifndef _CONCURRENTIOTIVITYUTILS_H_
  24 #define _CONCURRENTIOTIVITYUTILS_H_
  25
  26 #include <queue>
  27 #include <mutex>
  28 #include <memory>
  29 #include <condition_variable>
  30 #include <thread>
  31 #include <unistd.h>
  32 #include <iostream>
  33 #include <string>
  34 #include <memory>
  35 #include <map>
  36 #include "IotivityWorkItem.h"
  37 #include "WorkQueue.h"
  38 #include "ocstack.h"
  39 #include "octypes.h"
  40
  41 namespace OC
  42 {
  43     namespace Bridging
  44     {
  45         template<typename T, typename... Args>
  46         std::unique_ptr<T> make_unique(Args &&... args)
  47         {
  48             return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
  49         }
  50
  51         /**
  52          * Provides a synchronized C++ wrapper over the Iotivity CSDK.
  53          * Accepts workItems from the plugins for common operations.
  54          * A consumer thread processes these worker items and makes calls into Iotivity.
  55          * Another thread calls OCProcess() to check for network requests.
  56          */
  57         class ConcurrentIotivityUtils
  58         {
  59             private:
  60
  61                 static std::unique_ptr<WorkQueue<std::unique_ptr<IotivityWorkItem>>> m_queue;
  62                 std::mutex m_iotivityApiCallMutex;
  63
  64                 std::thread m_processWorkQueueThread, m_ocProcessThread;
  65                 bool m_threadStarted;
  66                 bool m_shutDownOCProcessThread;
  67                 static const int OCPROCESS_SLEEP_MICROSECONDS = 200000;
  68
  69                 // Fetches work item from queue and processes it.
  70                 void processWorkQueue()
  71                 {
  72                     while (true)
  73                     {
  74                         std::unique_ptr<IotivityWorkItem> workItem;
  75                         bool fetchedWorkItem = m_queue->get(&workItem);
  76
  77                         if (fetchedWorkItem)
  78                         {
  79                             std::lock_guard<std::mutex> lock(m_iotivityApiCallMutex);
  80                             workItem->process();
  81                         }
  82                         else
  83                         {
  84                             break;
  85                         }
  86                     }
  87                 }
  88
  89                 void callOCProcess()
  90                 {
  91                     while (!m_shutDownOCProcessThread)
  92                     {
  93                         {
  94                             std::lock_guard<std::mutex> lock(m_iotivityApiCallMutex);
  95                             OCProcess();
  96                         }
  97                         // Hopefully it's enough for other threads to be scheduled
  98                         // instead of the spin here. OCProcess is very lightweight though.
  99                         usleep(OCPROCESS_SLEEP_MICROSECONDS);
 100                     }
 101                 }
 102
 103             public:
 104
 105
 106                 ConcurrentIotivityUtils(std::unique_ptr<WorkQueue<std::unique_ptr<IotivityWorkItem>>>
 107                                         queueToMonitor)
 108                 {
 109                     m_queue = std::move(queueToMonitor);
 110                     m_threadStarted = false;
 111                     m_shutDownOCProcessThread = false;
 112                 }
 113
 114                 /**
 115                  * Starts 2 worker threads. One to service the concurrent work queue to call
 116                  * into Iotivity. One to process network requests by calling OCProcess()
 117                  */
 118                 void startWorkerThreads();
 119
 120                 /**
 121                  * Stops the 2 worker threads started by startWorkerThreads. @see startWorkerThreads
 122                  */
 123                 void stopWorkerThreads();
 124
 125                 /**
 126                  * Gets the string uri associated with an Iotivity handle.
 127                  * @warning This function is not thread safe and should only be called from entityHandler
 128                  *          specified when creating the resource. The entityhandler is called with the
 129                  *          Iotivity access mutex locked and this function does not modify anything
 130                  *          in the stack.
 131                  *
 132                  * @param[in]  handle    handle for the resource
 133                  * @param[out] uri       uri associated with the handle
 134                  *
 135                  * @return true if the resource is found and uri will be populated, else false.
 136                  */
 137                 bool static getUriFromHandle(OCResourceHandle handle, std::string &uri);
 138
 139                 /**
 140                  * Sends out OBSERVE notifications for the resource with the given uri.
 141                  * Notifications are sent out using OC_NA_QOS.
 142                  *
 143                  * @param[in] resourceUri      resource uri for fetching handle and notifying
 144                  *
 145                  * @return OCStackResult OC_STACK_OK on success, some other value upon failure.
 146                  */
 147                 OCStackResult static queueNotifyObservers(const std::string &resourceUri);
 148
 149                 /**
 150                  * Create an Iotivity resource with the given properties.
 151                  *
 152                  * @param[in] uri
 153                  * @param[in] resourceType
 154                  * @param[in] interface
 155                  * @param[in] entityHandler  Callback function that will be called on requests to this resource.
 156                  * @param[in] callbackParam
 157                  * @param[in] resourceProperties
 158                  *
 159                  * @return OCStackResult OC_STACK_OK on success, some other value upon failure.
 160                  */
 161                 OCStackResult static
 162                 queueCreateResource(const std::string &uri, const std::string &resourceType,
 163                                const std::string &interface,
 164                                OCEntityHandler entityHandler, void *callbackParam,
 165                                uint8_t resourceProperties);
 166
 167                 /**
 168                  * Delete the Iotivity resource given in the uri.
 169                  *
 170                  * @param[in] uri
 171                  *
 172                  * @return OCStackResult OC_STACK_OK on success, some other value upon failure.
 173                  */
 174                 OCStackResult static queueDeleteResource(const std::string &uri);
 175
 176                 /**
 177                  * Send a response to a request.
 178                  *
 179                  * @param[in] request OCEntityHandleRequest type that was handed in the entityhandler.
 180                  * @param[in] payload The response payload. This is cloned and callee still has ownership
 181                  *                and the onus to free this.
 182                  * @param[in] responseCode The response code of type OCEntityHandlerResult in ocstack.h
 183                  *
 184                  * @return OCStackResult OC_STACK_OK on success, some other value upon failure.
 185                  */
 186                 OCStackResult static
 187                 respondToRequest(OCEntityHandlerRequest *request, OCRepPayload *payload,
 188                                  OCEntityHandlerResult responseCode);
 189
 190                 /**
 191                  * Respond with an error message. Internally calls
 192                  * ConcurrentIotivityUtils::respondToRequest() after creating
 193                  * a payload containing the error message. The error message
 194                  * key will be x.org.iotivity.error and the value will be
 195                  * errorMessage.
 196                  *
 197                  * @param[in] request EntityHandler request
 198                  * @param[in] errorMessage May be NULL.
 199                  * @param[in] errorCode entity handler result
 200                  *
 201                  * @return OCStackResult OC_STACK_OK on success, some other value upon failure.
 202                  */
 203                 OCStackResult static respondToRequestWithError(OCEntityHandlerRequest *request,
 204                         const std::string &errorMessage,
 205                         OCEntityHandlerResult errorCode);
 206
 207                 /**
 208                  * Parse the query parameter as a keyValueMap
 209                  *
 210                  * @param[in] query query to be parsed
 211                  * @param[in,out] keyValueMap key value map of the query
 212                  */
 213                 void static getKeyValueParams(const std::string &query,
 214                                               std::map<std::string, std::string> &keyValueMap);
 215
 216                 /**
 217                  * Can be called from the entity handler to handle requests for default interface
 218                  *
 219                  * @param[in] query
 220                  *
 221                  * @return true if request for default interface (oic.if.baseline) else false.
 222                  */
 223                 bool static isRequestForDefaultInterface(const std::string &query);
 224
 225         };
 226     }
 227 }
 228
 229 #endif //_CONCURRENTIOTIVITYUTILS_H_