MQTTSNGWProcess.cpp

00001 /**************************************************************************************
00002  * Copyright (c) 2016, Tomoaki Yamaguchi
00003  *
00004  * All rights reserved. This program and the accompanying materials
00005  * are made available under the terms of the Eclipse Public License v1.0
00006  * and Eclipse Distribution License v1.0 which accompany this distribution.
00007  *
00008  * The Eclipse Public License is available at
00009  *    http://www.eclipse.org/legal/epl-v10.html
00010  * and the Eclipse Distribution License is available at
00011  *   http://www.eclipse.org/org/documents/edl-v10.php.
00012  *
00013  * Contributors:
00014  *    Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
00015  **************************************************************************************/
00016 
00017 #include <stdio.h>
00018 #include <stdlib.h>
00019 #include <string.h>
00020 #include <stdarg.h>
00021 #include <signal.h>
00022 #include <Timer.h>
00023 #include <exception>
00024 #include <getopt.h>
00025 #include <unistd.h>
00026 #include "MQTTSNGWProcess.h"
00027 #include "Threading.h"
00028 
00029 using namespace std;
00030 using namespace MQTTSNGW;
00031 
00032 char* currentDateTime(void);
00033 
00034 /*=====================================
00035  Global Variables & Functions
00036  ======================================*/
00037 Process* MQTTSNGW::theProcess = nullptr;
00038 MultiTaskProcess* MQTTSNGW::theMultiTaskProcess = nullptr;
00039 
00040 /*
00041  *  Save the type of signal
00042  */
00043 volatile int theSignaled = 0;
00044 
00045 static void signalHandler(int sig)
00046 {
00047     theSignaled = sig;
00048 }
00049 
00050 /*=====================================
00051  Class Process
00052  ====================================*/
00053 Process::Process()
00054 {
00055     _argc = 0;
00056     _argv = 0;
00057     _configDir = CONFIG_DIRECTORY;
00058     _configFile = CONFIG_FILE;
00059     _log = 0;
00060 }
00061 
00062 Process::~Process()
00063 {
00064     if (_rb )
00065     {
00066         delete _rb;
00067     }
00068     if ( _rbsem )
00069     {
00070         delete _rbsem;
00071     }
00072 }
00073 
00074 void Process::run()
00075 {
00076 
00077 }
00078 
00079 void Process::initialize(int argc, char** argv)
00080 {
00081     char param[MQTTSNGW_PARAM_MAX];
00082     _argc = argc;
00083     _argv = argv;
00084     signal(SIGINT, signalHandler);
00085     signal(SIGTERM, signalHandler);
00086     signal(SIGHUP, signalHandler);
00087 
00088     int opt;
00089     while ((opt = getopt(_argc, _argv, "f:")) != -1)
00090     {
00091         if ( opt == 'f' )
00092         {
00093             string config = string(optarg);
00094             size_t pos = 0;
00095             if ( (pos = config.find_last_of("/")) == string::npos )
00096             {
00097                 _configFile = optarg;
00098             }
00099             else
00100             {
00101                 _configFile = config.substr(pos + 1, config.size() - pos - 1);;
00102                 _configDir = config.substr(0, pos + 1);
00103             }
00104         }
00105     }
00106     _rbsem = new Semaphore(MQTTSNGW_RB_SEMAPHOR_NAME, 0);
00107     _rb = new RingBuffer(_configDir.c_str());
00108 
00109     if (getParam("ShearedMemory", param) == 0)
00110     {
00111         if (!strcasecmp(param, "YES"))
00112         {
00113             _log = 1;
00114         }
00115         else
00116         {
00117             _log = 0;
00118         }
00119     }
00120 }
00121 
00122 void Process::putLog(const char* format, ...)
00123 {
00124     _mt.lock();
00125     va_list arg;
00126     va_start(arg, format);
00127     vsprintf(_rbdata, format, arg);
00128     va_end(arg);
00129     if (strlen(_rbdata))
00130     {
00131         if ( _log > 0 )
00132         {
00133             _rb->put(_rbdata);
00134             _rbsem->post();
00135         }
00136         else
00137         {
00138             printf("%s", _rbdata);
00139         }
00140     }
00141     _mt.unlock();
00142 }
00143 
00144 int Process::getArgc()
00145 {
00146     return _argc;
00147 }
00148 
00149 char** Process::getArgv()
00150 {
00151     return _argv;
00152 }
00153 
00154 int Process::getParam(const char* parameter, char* value)
00155 {
00156     char str[MQTTSNGW_PARAM_MAX];
00157     char param[MQTTSNGW_PARAM_MAX];
00158     FILE *fp;
00159 
00160     int i = 0, j = 0;
00161     string configPath = _configDir + _configFile;
00162 
00163     if ((fp = fopen(configPath.c_str(), "r")) == NULL)
00164     {
00165         WRITELOG("No config file:[%s]\n", configPath.c_str());
00166         return -1;
00167     }
00168 
00169     while (true)
00170     {
00171         if (fgets(str, MQTTSNGW_PARAM_MAX - 1, fp) == NULL)
00172         {
00173             fclose(fp);
00174             return -3;
00175         }
00176         if (!strncmp(str, parameter, strlen(parameter)))
00177         {
00178             while (str[i++] != '=')
00179             {
00180                 ;
00181             }
00182             while (str[i] != '\n')
00183             {
00184                 param[j++] = str[i++];
00185             }
00186             param[j] = '\0';
00187 
00188             for (i = strlen(param) - 1; i >= 0 && isspace(param[i]); i--)
00189                 ;
00190             param[i + 1] = '\0';
00191             for (i = 0; isspace(param[i]); i++)
00192                 ;
00193             if (i > 0)
00194             {
00195                 j = 0;
00196                 while (param[i])
00197                     param[j++] = param[i++];
00198                 param[j] = '\0';
00199             }
00200             strcpy(value, param);
00201             fclose(fp);
00202             return 0;
00203         }
00204     }
00205     fclose(fp);
00206     return -2;
00207 }
00208 
00209 const char* Process::getLog()
00210 {
00211     int len = 0;
00212     _mt.lock();
00213     while ((len = _rb->get(_rbdata, PROCESS_LOG_BUFFER_SIZE)) == 0)
00214     {
00215         _rbsem->timedwait(1000);
00216         if ( checkSignal() == SIGINT)
00217         {
00218             break;
00219         }
00220     }
00221     *(_rbdata + len) = 0;
00222     _mt.unlock();
00223     return _rbdata;
00224 }
00225 
00226 void Process::resetRingBuffer()
00227 {
00228     _rb->reset();
00229 }
00230 
00231 int Process::checkSignal(void)
00232 {
00233     return theSignaled;
00234 }
00235 
00236 const string* Process::getConfigDirName(void)
00237 {
00238     return &_configDir;
00239 }
00240 
00241 const string* Process::getConfigFileName(void)
00242 {
00243     return &_configFile;
00244 }
00245 
00246 /*=====================================
00247  Class MultiTaskProcess
00248  ====================================*/
00249 MultiTaskProcess::MultiTaskProcess()
00250 {
00251     theMultiTaskProcess = this;
00252     _threadCount = 0;
00253     _stopCount = 0;
00254 }
00255 
00256 MultiTaskProcess::~MultiTaskProcess()
00257 {
00258     for (int i = 0; i < _threadCount; i++)
00259     {
00260         _threadList[i]->stop();
00261     }
00262 }
00263 
00264 void MultiTaskProcess::initialize(int argc, char** argv)
00265 {
00266     Process::initialize(argc, argv);
00267     for (int i = 0; i < _threadCount; i++)
00268     {
00269         _threadList[i]->initialize(argc, argv);
00270     }
00271 
00272 }
00273 
00274 void MultiTaskProcess::run(void)
00275 {
00276     for (int i = 0; i < _threadCount; i++)
00277     {
00278         _threadList[i]->start();
00279     }
00280 
00281     try
00282     {
00283         while(true)
00284         {
00285             if (theProcess->checkSignal() == SIGINT)
00286             {
00287                 return;
00288             }
00289             sleep(1);
00290         }
00291     }
00292     catch(Exception* ex)
00293     {
00294         ex->writeMessage();
00295     }
00296     catch(...)
00297     {
00298         throw;
00299     }
00300 }
00301 
00302 void MultiTaskProcess::waitStop(void)
00303 {
00304     while (_stopCount < _threadCount)
00305     {
00306         sleep(1);
00307     }
00308 }
00309 
00310 void MultiTaskProcess::threadStoped(void)
00311 {
00312     _mutex.lock();
00313     _stopCount++;
00314     _mutex.unlock();
00315 
00316 }
00317 
00318 void MultiTaskProcess::attach(Thread* thread)
00319 {
00320     _mutex.lock();
00321     if (_threadCount < MQTTSNGW_MAX_TASK)
00322     {
00323         _threadList[_threadCount] = thread;
00324         _threadCount++;
00325     }
00326     else
00327     {
00328         _mutex.unlock();
00329         throw Exception("Full of Threads");
00330     }
00331     _mutex.unlock();
00332 }
00333 
00334 int MultiTaskProcess::getParam(const char* parameter, char* value)
00335 {
00336     _mutex.lock();
00337     int rc = Process::getParam(parameter, value);
00338     _mutex.unlock();
00339     if (rc == -1)
00340     {
00341         throw Exception("No config file.");
00342     }
00343     return rc;
00344 }
00345 
00346 /*=====================================
00347  Class Exception
00348  ======================================*/
00349 Exception::Exception(const string& message)
00350 {
00351     _message = message;
00352     _exNo = 0;
00353     _fileName = 0;
00354     _functionName = 0;
00355     _line = 0;
00356 }
00357 
00358 Exception::Exception(const int exNo, const string& message)
00359 {
00360     _message = message;
00361     _exNo = exNo;
00362     _fileName = nullptr;
00363     _functionName = nullptr;
00364     _line = 0;
00365 }
00366 
00367 Exception::Exception(const int exNo, const string& message, const char* file,
00368         const char* function, const int line)
00369 {
00370     _message = message;
00371     _exNo = exNo;
00372     _fileName = file;
00373     _functionName = function;
00374     _line = line;
00375 }
00376 
00377 Exception::~Exception() throw ()
00378 {
00379 
00380 }
00381 
00382 const char* Exception::what() const throw ()
00383 {
00384     return _message.c_str();
00385 }
00386 
00387 const char* Exception::getFileName()
00388 {
00389     return _fileName;
00390 }
00391 
00392 const char* Exception::getFunctionName()
00393 {
00394     return _functionName;
00395 }
00396 
00397 const int Exception::getLineNo()
00398 {
00399     return _line;
00400 }
00401 
00402 const int Exception::getExceptionNo()
00403 {
00404     return _exNo;
00405 }
00406 
00407 void Exception::writeMessage()
00408 {
00409     if (getExceptionNo() == 0 )
00410     {
00411         WRITELOG("%s %s\n", currentDateTime(), what());
00412     }
00413     else
00414     {
00415         WRITELOG("%s:%-6d   %s  line %-4d %s() : %s\n", currentDateTime(), getExceptionNo(),
00416             getFileName(), getLineNo(), getFunctionName(), what());
00417     }
00418 }