PluginRoutingInterfaceAsync/include/CAmRoutingSenderAsync.h

   1 /**
   2  *  Copyright (c) 2012 BMW
   3  *
   4  *  \author Christian Mueller, christian.ei.mueller@bmw.de BMW 2011,2012
   5  *
   6  *  \copyright
   7  *  Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction,
   8  *  including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so,
   9  *  subject to the following conditions:
  10  *  The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
  11  *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
  12  *  IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR
  13  *  THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  14  *
  15  *  For further information see http://www.genivi.org/.
  16  */
  17
  18 #ifndef ROUTINGSENDER_H_
  19 #define ROUTINGSENDER_H_
  20
  21 #include "routing/IAmRoutingSend.h"
  22 #include "IAmRoutingReceiverShadow.h"
  23 #include <semaphore.h>
  24 #include <memory.h>
  25
  26 namespace am
  27 {
  28
  29 class CAmWorkerThreadPool;
  30
  31 /**
  32  * Base class for workers implements everything that is needed to implement a workerthread
  33  * inherit from this class when adding new workers
  34  */
  35 class CAmWorker
  36 {
  37 public:
  38     CAmWorker(CAmWorkerThreadPool* pool);
  39     virtual ~CAmWorker()
  40     {};
  41     /**
  42      * needs to be overwritten, this function is called when the worker should start to work
  43      */
  44     void virtual start2work()=0;
  45
  46     /**
  47      * needs to be overwritten, this function is called when the worker thread is canceled. Should be used for
  48      * clean up and sending important messages
  49      */
  50     void virtual cancelWork()=0;
  51     /**
  52      * waits for a semaphore with a timeout. This leaves the Threadpool the chance to interrupt the processing
  53      * You should call whenever waiting on some event, even with time=0 in order to make sure that cancel events are
  54      * received
  55      * @param time time until timeout in timespec format
  56      * @return true if thread is canceled. Then just return start2work function so that the thread is given back to the pool
  57      */
  58     bool timedWait(timespec time);
  59
  60     /**
  61      * the semaphore for cancellation is set by the thread automatically...
  62      * @param cancel
  63      */
  64     void setCancelSempaphore(sem_t* cancel);
  65     CAmWorkerThreadPool* pPool;
  66 private:
  67     sem_t* mCancelSem; //<! semaphore for cancellation
  68 };
  69
  70 /**
  71  * This class handles the threadpool
  72  */
  73 class CAmWorkerThreadPool
  74 {
  75 public:
  76     /**
  77      * creates the pool. Give the max number of threads as argument
  78      * @param numThreads max number of threads
  79      */
  80     CAmWorkerThreadPool(int numThreads);
  81     virtual ~CAmWorkerThreadPool();
  82
  83     /**
  84      * assigns a thread to a worker class and starts working.
  85      * @param worker
  86      * @return the actual assigned workerID or -1 in case no thread is free
  87      */
  88     int16_t startWork(CAmWorker* worker);
  89     /**
  90      * cancels a thread
  91      * @param workerID thread to be canceled
  92      * @return true if thread was found, false if not
  93      */
  94     bool cancelWork(int workerID);
  95
  96     /**
  97      * the workers call this function upon completion of their task
  98      * @param threadID
  99      */
 100     void finishedWork(pthread_t threadID);
 101
 102 private:
 103     static void* CAmWorkerThread(void* data);
 104     int mNumThreads;
 105     struct threadInfo_s
 106     {
 107         uint16_t workerID;
 108         pthread_t threadID;
 109         bool busy;
 110         sem_t block;
 111         sem_t cancel;
 112         CAmWorker *worker;
 113     };
 114     std::vector<threadInfo_s> mListWorkers; //<! list of all workers
 115     static pthread_mutex_t mBlockingMutex; //<! mutex to block the acces of the list
 116 };
 117
 118 class CAmRoutingSenderAsync: public IAmRoutingSend
 119 {
 120 public:
 121     CAmRoutingSenderAsync();
 122     virtual ~CAmRoutingSenderAsync();
 123     am_Error_e startupInterface(IAmRoutingReceive* routingreceiveinterface) ;
 124     void setRoutingReady(const uint16_t handle) ;
 125     void setRoutingRundown(const uint16_t handle) ;
 126     am_Error_e asyncAbort(const am_Handle_s handle) ;
 127     am_Error_e asyncConnect(const am_Handle_s handle, const am_connectionID_t connectionID, const am_sourceID_t sourceID, const am_sinkID_t sinkID, const am_ConnectionFormat_e connectionFormat) ;
 128     am_Error_e asyncDisconnect(const am_Handle_s handle, const am_connectionID_t connectionID) ;
 129     am_Error_e asyncSetSinkVolume(const am_Handle_s handle, const am_sinkID_t sinkID, const am_volume_t volume, const am_RampType_e ramp, const am_time_t time) ;
 130     am_Error_e asyncSetSourceVolume(const am_Handle_s handle, const am_sourceID_t sourceID, const am_volume_t volume, const am_RampType_e ramp, const am_time_t time) ;
 131     am_Error_e asyncSetSourceState(const am_Handle_s handle, const am_sourceID_t sourceID, const am_SourceState_e state) ;
 132     am_Error_e asyncSetSinkSoundProperties(const am_Handle_s handle, const am_sinkID_t sinkID, const std::vector<am_SoundProperty_s>& listSoundProperties) ;
 133     am_Error_e asyncSetSinkSoundProperty(const am_Handle_s handle, const am_sinkID_t sinkID, const am_SoundProperty_s& soundProperty) ;
 134     am_Error_e asyncSetSourceSoundProperties(const am_Handle_s handle, const am_sourceID_t sourceID, const std::vector<am_SoundProperty_s>& listSoundProperties) ;
 135     am_Error_e asyncSetSourceSoundProperty(const am_Handle_s handle, const am_sourceID_t sourceID, const am_SoundProperty_s& soundProperty) ;
 136     am_Error_e asyncCrossFade(const am_Handle_s handle, const am_crossfaderID_t crossfaderID, const am_HotSink_e hotSink, const am_RampType_e rampType, const am_time_t time) ;
 137     am_Error_e setDomainState(const am_domainID_t domainID, const am_DomainState_e domainState) ;
 138     am_Error_e returnBusName(std::string& BusName) const ;
 139     void getInterfaceVersion(std::string& version) const ;
 140
 141     /**
 142      * threadafe insert of route and connectionID
 143      * @param connectionID
 144      * @param route
 145      */
 146     void insertConnectionSafe(am_connectionID_t connectionID, am_RoutingElement_s route);
 147
 148     /**
 149      * threadsafe removal of a connection
 150      * @param
 151      */
 152     void removeConnectionSafe(am_connectionID_t);
 153
 154     /**
 155      * threadsafe remove of a handle
 156      * @param handle
 157      */
 158     void removeHandleSafe(uint16_t handle);
 159
 160     /**
 161      * threadsafe update of Sinkvolume
 162      * @param sinkID
 163      * @param volume
 164      */
 165     void updateSinkVolumeSafe(am_sinkID_t sinkID, am_volume_t volume);
 166
 167     /**
 168      * threadsafe update of SourceVolume
 169      * @param sourceID
 170      * @param volume
 171      */
 172     void updateSourceVolumeSafe(am_sourceID_t sourceID, am_volume_t volume);
 173
 174     /**
 175      * threadsafe update of sourceState
 176      * @param sourceID
 177      * @param state
 178      */
 179     void updateSourceStateSafe(am_sourceID_t sourceID, am_SourceState_e state);
 180
 181     /**
 182      * threadsafe update of sinkSoundProperty
 183      * @param sinkID
 184      * @param soundProperty
 185      */
 186     void updateSinkSoundPropertySafe(am_sinkID_t sinkID, am_SoundProperty_s soundProperty);
 187
 188     /**
 189      * threadsafe update of sourceSoundProperty
 190      * @param sourceID
 191      * @param soundProperty
 192      */
 193     void updateSourceSoundPropertySafe(am_sourceID_t sourceID, am_SoundProperty_s soundProperty);
 194
 195     /**
 196      * threadsafe update of domainstate
 197      * @param domainID
 198      * @param domainState
 199      */
 200     void updateDomainstateSafe(am_domainID_t domainID, am_DomainState_e domainState);
 201
 202     void updateDomainListSafe(std::vector<am_Domain_s> listDomains);
 203
 204     void updateSourceListSafe(std::vector<am_Source_s> listSource);
 205
 206     void updateSinkListSafe(std::vector<am_Sink_s> listSinks);
 207
 208 private:
 209     /**
 210      * Extra thread that handles dbus stimulation for interrupt tests
 211      * This is a very very very basic implementation of the dbus interface
 212      * there is not failure handling, nothing.
 213      * it is used just for testing, not intended to be used otherwise...
 214      * @param data
 215      */
 216     std::vector<am_Domain_s> createDomainTable();
 217     std::vector<am_Sink_s> createSinkTable();
 218     std::vector<am_Source_s> createSourceTable();
 219     std::vector<am_Gateway_s> createGatewayTable();
 220     IAmRoutingReceiverShadow* mShadow;
 221     IAmRoutingReceive* mReceiveInterface;
 222     CAmSocketHandler *mSocketHandler;
 223     std::vector<am_Domain_s> mDomains;
 224     std::vector<am_Sink_s> mSinks;
 225     std::vector<am_Source_s> mSources;
 226     std::vector<am_Gateway_s> mGateways;
 227     std::map<uint16_t, int16_t> mMapHandleWorker;
 228     std::map<am_connectionID_t, am_RoutingElement_s> mMapConnectionIDRoute;
 229     CAmWorkerThreadPool mPool;
 230     pthread_t mInterruptThread;
 231     static pthread_mutex_t mMapConnectionMutex;
 232     static pthread_mutex_t mMapHandleWorkerMutex;
 233     static pthread_mutex_t mSinksMutex;
 234     static pthread_mutex_t mSourcesMutex;
 235     static pthread_mutex_t mDomainsMutex;
 236 };
 237
 238 /**
 239  * worker to for connection
 240  */
 241 class asycConnectWorker: public CAmWorker
 242 {
 243 public:
 244     asycConnectWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_Handle_s handle, const am_connectionID_t connectionID, const am_sourceID_t sourceID, const am_sinkID_t sinkID, const am_ConnectionFormat_e connectionFormat);
 245     void start2work();
 246     void cancelWork();
 247 private:
 248     CAmRoutingSenderAsync * mAsyncSender;
 249     IAmRoutingReceiverShadow *mShadow;
 250     am_Handle_s mHandle;
 251     am_connectionID_t mConnectionID;
 252     am_sourceID_t mSourceID;
 253     am_sinkID_t mSinkID;
 254     am_ConnectionFormat_e mConnectionFormat;
 255 };
 256
 257 /**
 258  * worker for disconnecting
 259  */
 260 class asycDisConnectWorker: public CAmWorker
 261 {
 262 public:
 263     asycDisConnectWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_Handle_s handle, const am_connectionID_t connectionID);
 264     void start2work();
 265     void cancelWork();
 266 private:
 267     CAmRoutingSenderAsync * mAsyncSender;
 268     IAmRoutingReceiverShadow *mShadow;
 269     am_Handle_s mHandle;
 270     am_connectionID_t mConnectionID;
 271     am_ConnectionFormat_e mConnectionFormat;
 272 };
 273
 274 /**
 275  * worker to for connection
 276  */
 277
 278 #include <semaphore.h>
 279 #include <sys/signalfd.h>
 280 #include <signal.h>
 281
 282 class asyncSetSinkVolumeWorker: public CAmWorker
 283 {
 284 public:
 285     asyncSetSinkVolumeWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_volume_t currentVolume, const am_Handle_s handle, const am_sinkID_t sinkID, const am_volume_t volume, const am_RampType_e ramp, const am_time_t time);
 286     void start2work();
 287     void cancelWork();
 288 private:
 289     CAmRoutingSenderAsync * mAsyncSender;
 290     IAmRoutingReceiverShadow *mShadow;
 291     am_volume_t mCurrentVolume;
 292     am_Handle_s mHandle;
 293     am_sinkID_t mSinkID;
 294     am_volume_t mVolume;
 295     am_RampType_e mRamp;
 296     am_time_t mTime;
 297 };
 298
 299 class asyncSetSourceVolumeWorker: public CAmWorker
 300 {
 301 public:
 302     asyncSetSourceVolumeWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_volume_t currentVolume, const am_Handle_s handle, const am_sourceID_t SourceID, const am_volume_t volume, const am_RampType_e ramp, const am_time_t time);
 303     void start2work();
 304     void cancelWork();
 305 private:
 306     CAmRoutingSenderAsync * mAsyncSender;
 307     IAmRoutingReceiverShadow *mShadow;
 308     am_volume_t mCurrentVolume;
 309     am_Handle_s mHandle;
 310     am_sourceID_t mSourceID;
 311     am_volume_t mVolume;
 312     am_RampType_e mRamp;
 313     am_time_t mTime;
 314 };
 315
 316 class asyncSetSourceStateWorker: public CAmWorker
 317 {
 318 public:
 319     asyncSetSourceStateWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_Handle_s handle, const am_sourceID_t sourceID, const am_SourceState_e state);
 320     void start2work();
 321     void cancelWork();
 322 private:
 323     CAmRoutingSenderAsync * mAsyncSender;
 324     IAmRoutingReceiverShadow *mShadow;
 325     am_Handle_s mHandle;
 326     am_sourceID_t mSourceID;
 327     am_SourceState_e mSourcestate;
 328 };
 329
 330 class asyncSetSinkSoundPropertyWorker: public CAmWorker
 331 {
 332 public:
 333     asyncSetSinkSoundPropertyWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_Handle_s handle, const am_SoundProperty_s soundProperty, const am_sinkID_t sinkID);
 334     void start2work();
 335     void cancelWork();
 336 private:
 337     CAmRoutingSenderAsync * mAsyncSender;
 338     IAmRoutingReceiverShadow *mShadow;
 339     am_Handle_s mHandle;
 340     am_sinkID_t mSinkID;
 341     am_SoundProperty_s mSoundProperty;
 342 };
 343
 344 class asyncSetSourceSoundPropertyWorker: public CAmWorker
 345 {
 346 public:
 347     asyncSetSourceSoundPropertyWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_Handle_s handle, const am_SoundProperty_s soundProperty, const am_sourceID_t sourceID);
 348     void start2work();
 349     void cancelWork();
 350 private:
 351     CAmRoutingSenderAsync * mAsyncSender;
 352     IAmRoutingReceiverShadow *mShadow;
 353     am_Handle_s mHandle;
 354     am_sourceID_t mSourceID;
 355     am_SoundProperty_s mSoundProperty;
 356 };
 357
 358 class asyncDomainStateChangeWorker: public CAmWorker
 359 {
 360 public:
 361     asyncDomainStateChangeWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const am_domainID_t domainID, const am_DomainState_e domainState);
 362     void start2work();
 363     void cancelWork();
 364 private:
 365     CAmRoutingSenderAsync * mAsyncSender;
 366     IAmRoutingReceiverShadow *mShadow;
 367     am_domainID_t mDomainID;
 368     am_DomainState_e mDomainState;
 369 };
 370
 371 class syncRegisterWorker: public CAmWorker
 372 {
 373 public:
 374     syncRegisterWorker(CAmRoutingSenderAsync * asyncSender, CAmWorkerThreadPool* pool, IAmRoutingReceiverShadow* shadow, const std::vector<am_Domain_s> domains, const std::vector<am_Sink_s> sinks, const std::vector<am_Source_s> sources, const  uint16_t handle);
 375     void start2work();
 376     void cancelWork();
 377 private:
 378     CAmRoutingSenderAsync * mAsyncSender;
 379     IAmRoutingReceiverShadow *mShadow;
 380     std::vector<am_Domain_s> mListDomains;
 381     std::vector<am_Sink_s> mListSinks;
 382     std::vector<am_Source_s> mListSources;
 383     uint16_t mHandle;
 384 };
 385
 386 }
 387
 388 #endif /* ROUTINGSENDER_H_ */