source: ntrip/trunk/BNC/src/bnccaster.cpp @ 8920

Last change on this file since 8920 was 8920, checked in by stuerze, 6 months ago

minor changes

File size: 17.0 KB
Line 
1// Part of BNC, a utility for retrieving decoding and
2// converting GNSS data streams from NTRIP broadcasters.
3//
4// Copyright (C) 2007
5// German Federal Agency for Cartography and Geodesy (BKG)
6// http://www.bkg.bund.de
7// Czech Technical University Prague, Department of Geodesy
8// http://www.fsv.cvut.cz
9//
10// Email: euref-ip@bkg.bund.de
11//
12// This program is free software; you can redistribute it and/or
13// modify it under the terms of the GNU General Public License
14// as published by the Free Software Foundation, version 2.
15//
16// This program is distributed in the hope that it will be useful,
17// but WITHOUT ANY WARRANTY; without even the implied warranty of
18// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19// GNU General Public License for more details.
20//
21// You should have received a copy of the GNU General Public License
22// along with this program; if not, write to the Free Software
23// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
24
25/* -------------------------------------------------------------------------
26 * BKG NTRIP Client
27 * -------------------------------------------------------------------------
28 *
29 * Class:      bncCaster
30 *
31 * Purpose:    buffers and disseminates the data
32 *
33 * Author:     L. Mervart
34 *
35 * Created:    24-Dec-2005
36 *
37 * Changes:
38 *
39 * -----------------------------------------------------------------------*/
40
41#include <math.h>
42#include <unistd.h>
43#include <iostream>
44#include <iomanip>
45#include <sstream>
46
47#include "bnccaster.h"
48#include "bncrinex.h"
49#include "bnccore.h"
50#include "bncgetthread.h"
51#include "bncutils.h"
52#include "bncsettings.h"
53
54using namespace std;
55
56// Constructor
57////////////////////////////////////////////////////////////////////////////
58bncCaster::bncCaster() {
59
60  bncSettings settings;
61
62  connect(this, SIGNAL(newMessage(QByteArray,bool)),
63          BNC_CORE, SLOT(slotMessage(const QByteArray,bool)));
64
65  _outFile = 0;
66  _out     = 0;
67  reopenOutFile();
68
69  int port = settings.value("outPort").toInt();
70
71  if (port != 0) {
72    _server = new QTcpServer;
73    if ( !_server->listen(QHostAddress::Any, port) ) {
74      QString message = "bncCaster: Cannot listen on sync port "
75                      + QByteArray::number(port)
76                      + " (Error Code: " + QByteArray::number(_server->serverError()) + ")";
77      emit newMessage(message.toLatin1(), true);
78    }
79    connect(_server, SIGNAL(newConnection()), this, SLOT(slotNewConnection()));
80    _sockets = new QList<QTcpSocket*>;
81  }
82  else {
83    _server  = 0;
84    _sockets = 0;
85  }
86
87  int uPort = settings.value("outUPort").toInt();
88  if (uPort != 0) {
89    _uServer = new QTcpServer;
90    if ( !_uServer->listen(QHostAddress::Any, uPort) ) {
91      QString message = "bncCaster: Cannot listen on usync port "
92                      + QByteArray::number(uPort)
93                      + " (Error Code: " + QByteArray::number(_uServer->serverError()) + ")";
94      emit newMessage(message.toLatin1(), true);
95    }
96    connect(_uServer, SIGNAL(newConnection()), this, SLOT(slotNewUConnection()));
97    _uSockets = new QList<QTcpSocket*>;
98  }
99  else {
100    _uServer  = 0;
101    _uSockets = 0;
102  }
103
104  _outLockTime = settings.value("outLockTime",false).toBool();
105  _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
106  _outWait            = settings.value("outWait").toInt();
107  if (_outWait < 1) {
108    _outWait = 1;
109  }
110  _confInterval = -1;
111
112  // Miscellaneous output port
113  // -------------------------
114  _miscMount = settings.value("miscMount").toString();
115  _miscPort  = settings.value("miscPort").toInt();
116  if (!_miscMount.isEmpty() && _miscPort != 0) {
117    _miscServer = new QTcpServer;
118    if ( !_miscServer->listen(QHostAddress::Any, _miscPort) ) {
119      QString message = "bncCaster: Cannot listen on miscellaneous output port "
120                      + QByteArray::number(_miscPort)
121                      + " (Error Code: " + QByteArray::number(_miscServer->serverError()) + ")";
122      emit newMessage(message.toLatin1(), true);
123    }
124    connect(_miscServer, SIGNAL(newConnection()), this, SLOT(slotNewMiscConnection()));
125    _miscSockets = new QList<QTcpSocket*>;
126  }
127  else {
128    _miscServer  = 0;
129    _miscSockets = 0;
130  }
131}
132
133// Destructor
134////////////////////////////////////////////////////////////////////////////
135bncCaster::~bncCaster() {
136
137  QMutexLocker locker(&_mutex);
138
139  QListIterator<bncGetThread*> it(_threads);
140  while(it.hasNext()){
141    bncGetThread* thread = it.next();
142    disconnect(thread, 0, 0, 0);
143    _staIDs.removeAll(thread->staID());
144    _threads.removeAll(thread);
145    thread->terminate();
146  }
147  delete _out;
148  delete _outFile;
149  delete _server;
150  delete _sockets;
151  delete _uServer;
152  delete _uSockets;
153  delete _miscServer;
154  delete _miscSockets;
155}
156
157// New Observations
158////////////////////////////////////////////////////////////////////////////
159void bncCaster::slotNewObs(const QByteArray staID, QList<t_satObs> obsList) {
160
161  QMutexLocker locker(&_mutex);
162
163  reopenOutFile();
164
165  unsigned index = 0;
166  QMutableListIterator<t_satObs> it(obsList);
167  while (it.hasNext()) {
168    ++index;
169    t_satObs& obs = it.next();
170
171    // Rename the Station
172    // ------------------
173    obs._staID = staID.data();
174
175    // Output into the socket
176    // ----------------------
177    if (_uSockets) {
178
179      ostringstream oStr;
180      oStr.setf(ios::showpoint | ios::fixed);
181      oStr << obs._staID                                        << " "
182           << setw(4)  << obs._time.gpsw()                      << " "
183           << setw(14) << setprecision(7) << obs._time.gpssec() << " "
184           << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
185
186      string hlpStr = oStr.str();
187
188      QMutableListIterator<QTcpSocket*> is(*_uSockets);
189      while (is.hasNext()) {
190        QTcpSocket* sock = is.next();
191        if (sock->state() == QAbstractSocket::ConnectedState) {
192          int numBytes = hlpStr.length();
193          if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
194            delete sock;
195            is.remove();
196          }
197        }
198        else if (sock->state() != QAbstractSocket::ConnectingState) {
199          delete sock;
200          is.remove();
201        }
202      }
203    }
204
205    // First time: set the _lastDumpTime
206    // ---------------------------------
207    if (!_lastDumpTime.valid()) {
208      _lastDumpTime = obs._time - 1.0;
209    }
210
211    // An old observation - throw it away
212    // ----------------------------------
213    if (obs._time <= _lastDumpTime) {
214      if (index == 1) {
215        bncSettings settings;
216        if ( !settings.value("outFile").toString().isEmpty() ||
217             !settings.value("outPort").toString().isEmpty() ) {
218          emit( newMessage(QString("%1: Old or erroneous observation epoch %2 thrown away from %3")
219                           .arg(staID.data())
220                           .arg(string(obs._time).c_str())
221                           .arg(obs._prn.toString().c_str())
222                             .toLatin1(), true) );
223        }
224      }
225      continue;
226    }
227
228    // Save the observation
229    // --------------------
230    _epochs[obs._time].append(obs);
231
232    // Dump Epochs
233    // -----------
234    double outWait = (double)_outWait;
235    if ((obs._time - outWait) > _lastDumpTime) {
236      dumpEpochs(obs._time - outWait);
237      _lastDumpTime = obs._time - outWait;
238    }
239  }
240}
241
242// New Connection
243////////////////////////////////////////////////////////////////////////////
244void bncCaster::slotNewConnection() {
245  _sockets->push_back( _server->nextPendingConnection() );
246  emit( newMessage(QString("New client connection on sync port: # %1")
247                   .arg(_sockets->size()).toLatin1(), true) );
248}
249
250void bncCaster::slotNewUConnection() {
251  _uSockets->push_back( _uServer->nextPendingConnection() );
252  emit( newMessage(QString("New client connection on usync port: # %1")
253                   .arg(_uSockets->size()).toLatin1(), true) );
254}
255
256// Add New Thread
257////////////////////////////////////////////////////////////////////////////
258void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
259
260  qRegisterMetaType<t_satObs>("t_satObs");
261  qRegisterMetaType< QList<t_satObs> >("QList<t_satObs>");
262
263  connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
264          this,      SLOT(slotNewObs(QByteArray, QList<t_satObs>)));
265
266  connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
267          this,      SIGNAL(newObs(QByteArray, QList<t_satObs>)));
268
269  connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
270          this,      SLOT(slotNewRawData(QByteArray, QByteArray)));
271
272  connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
273          this, SLOT(slotGetThreadFinished(QByteArray)));
274
275  _staIDs.push_back(getThread->staID());
276  _threads.push_back(getThread);
277
278  if (noNewThread) {
279    getThread->run();
280  }
281  else {
282    getThread->start();
283  }
284}
285
286// Get Thread destroyed
287////////////////////////////////////////////////////////////////////////////
288void bncCaster::slotGetThreadFinished(QByteArray staID) {
289  QMutexLocker locker(&_mutex);
290
291  QListIterator<bncGetThread*> it(_threads);
292  while (it.hasNext()) {
293    bncGetThread* thread = it.next();
294    if (thread->staID() == staID) {
295      _threads.removeOne(thread);
296    }
297  }
298
299  _staIDs.removeAll(staID);
300  emit( newMessage(
301           QString("Decoding %1 stream(s)").arg(_staIDs.size()).toLatin1(), true) );
302  if (_staIDs.size() == 0) {
303    emit(newMessage("bncCaster: Last get thread terminated", true));
304    emit getThreadsFinished();
305  }
306}
307
308// Dump Complete Epochs
309////////////////////////////////////////////////////////////////////////////
310void bncCaster::dumpEpochs(const bncTime& maxTime) {
311
312  QMutableMapIterator<bncTime, QList<t_satObs> > itEpo(_epochs);
313  while (itEpo.hasNext()) {
314    itEpo.next();
315    const bncTime& epoTime = itEpo.key();
316    if (epoTime <= maxTime) {
317      const QList<t_satObs>& allObs = itEpo.value();
318      int sec = int(nint(epoTime.gpssec()*10));
319      if ( (_out || _sockets) && (sec % (_samplingRateMult10) == 0) ) {
320        QListIterator<t_satObs> it(allObs);
321        bool firstObs = true;
322        while (it.hasNext()) {
323          const t_satObs& obs = it.next();
324
325          ostringstream oStr;
326          oStr.setf(ios::showpoint | ios::fixed);
327          if (firstObs) {
328            firstObs = false;
329            oStr << "> " << obs._time.gpsw() << ' '
330                 << setprecision(7) << obs._time.gpssec() << endl;
331          }
332          oStr << obs._staID << ' '
333               << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
334          if (!it.hasNext()) {
335            oStr << endl;
336          }
337          string hlpStr = oStr.str();
338
339          // Output into the File
340          // --------------------
341          if (_out) {
342            *_out << hlpStr.c_str();
343            _out->flush();
344          }
345
346          // Output into the socket
347          // ----------------------
348          if (_sockets) {
349            QMutableListIterator<QTcpSocket*> is(*_sockets);
350            while (is.hasNext()) {
351              QTcpSocket* sock = is.next();
352              if (sock->state() == QAbstractSocket::ConnectedState) {
353                int numBytes = hlpStr.length();
354                if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
355                  delete sock;
356                  is.remove();
357                }
358              }
359              else if (sock->state() != QAbstractSocket::ConnectingState) {
360                delete sock;
361                is.remove();
362              }
363            }
364          }
365        }
366      }
367      _epochs.remove(epoTime);
368    }
369  }
370}
371
372// Reread configuration (private slot)
373////////////////////////////////////////////////////////////////////////////
374void bncCaster::slotReadMountPoints() {
375
376  bncSettings settings;
377  settings.reRead();
378
379  readMountPoints();
380}
381
382// Read Mountpoints
383////////////////////////////////////////////////////////////////////////////
384void bncCaster::readMountPoints() {
385
386  bncSettings settings;
387
388  // Reread several options
389  // ----------------------
390  _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
391  _outWait      = settings.value("outWait").toInt();
392  if (_outWait < 1) {
393    _outWait = 1;
394  }
395
396  // Add new mountpoints
397  // -------------------
398  int iMount = -1;
399  QListIterator<QString> it(settings.value("mountPoints").toStringList());
400  while (it.hasNext()) {
401    ++iMount;
402    QStringList hlp = it.next().split(" ");
403    if (hlp.size() < 7) continue;
404    QUrl url(hlp[0]);
405
406    // Does it already exist?
407    // ----------------------
408    bool existFlg = false;
409    QListIterator<bncGetThread*> iTh(_threads);
410    while (iTh.hasNext()) {
411      bncGetThread* thread = iTh.next();
412      if (thread->mountPoint() == url) {
413        existFlg = true;
414        break;
415      }
416    }
417
418    // New bncGetThread
419    // ----------------
420    if (!existFlg) {
421      QByteArray format    = hlp[1].toLatin1();
422      QByteArray latitude  = hlp[3].toLatin1();
423      QByteArray longitude = hlp[4].toLatin1();
424      QByteArray nmea      = hlp[5].toLatin1();
425      QByteArray ntripVersion = hlp[6].toLatin1();
426
427      bncGetThread* getThread = new bncGetThread(url, format, latitude,
428                                        longitude, nmea, ntripVersion);
429      addGetThread(getThread);
430
431    }
432  }
433
434  // Remove mountpoints
435  // ------------------
436  QListIterator<bncGetThread*> iTh(_threads);
437  while (iTh.hasNext()) {
438    bncGetThread* thread = iTh.next();
439
440    bool existFlg = false;
441    QListIterator<QString> it(settings.value("mountPoints").toStringList());
442    while (it.hasNext()) {
443      QStringList hlp = it.next().split(" ");
444      if (hlp.size() <= 1) continue;
445      QUrl url(hlp[0]);
446
447      if (thread->mountPoint() == url) {
448        existFlg = true;
449        break;
450      }
451    }
452
453    if (!existFlg) {
454      disconnect(thread, 0, 0, 0);
455      _staIDs.removeAll(thread->staID());
456      _threads.removeAll(thread);
457      thread->terminate();
458    }
459  }
460
461  emit mountPointsRead(_threads);
462  emit( newMessage(QString("Configuration read: "
463                           + BNC_CORE->confFileName()
464                           + ", %1 stream(s)")
465                            .arg(_threads.count()).toLatin1(), true) );
466
467  // (Re-) Start the configuration timer
468  // -----------------------------------
469  if      (settings.value("onTheFlyInterval").toString() == "no") {
470    return;
471  }
472
473  int ms = 0;
474  if (_confInterval != -1) {
475    ms = 1000 * _confInterval;
476  }
477  else {
478    QTime currTime = currentDateAndTimeGPS().time();
479    QTime nextShotTime;
480    if      (settings.value("onTheFlyInterval").toString() == "1 min") {
481          _confInterval = 60;
482          nextShotTime = QTime(currTime.hour(), currTime.minute()+1, 0);
483    }
484    else if (settings.value("onTheFlyInterval").toString() == "5 min") {
485      _confInterval = 300;
486      nextShotTime = QTime(currTime.hour(), currTime.minute()+5, 0);
487    }
488    else if (settings.value("onTheFlyInterval").toString() == "1 hour") {
489      _confInterval = 3600;
490      nextShotTime = QTime(currTime.hour()+1, 0, 0);
491    }
492    else {
493      _confInterval = 86400;
494      nextShotTime = QTime(23, 59, 59, 999);
495    }
496
497    ms = currTime.msecsTo(nextShotTime);
498    if (ms < 30000) {
499      ms = 30000;
500    }
501  }
502
503  QTimer::singleShot(ms, this, SLOT(slotReadMountPoints()));
504}
505
506//
507////////////////////////////////////////////////////////////////////////////
508int bncCaster::myWrite(QTcpSocket* sock, const char* buf, int bufLen) {
509  sock->write(buf, bufLen);
510  for (int ii = 1; ii <= 10; ii++) {
511    if (sock->waitForBytesWritten(10)) {  // wait 10 ms
512      return bufLen;
513    }
514  }
515  return -1;
516}
517
518//
519////////////////////////////////////////////////////////////////////////////
520void bncCaster::reopenOutFile() {
521
522  bncSettings settings;
523
524  QString outFileName = settings.value("outFile").toString();
525  if ( !outFileName.isEmpty() ) {
526    expandEnvVar(outFileName);
527    if (!_outFile || _outFile->fileName() != outFileName) {
528      delete _out;
529      delete _outFile;
530      _outFile = new QFile(outFileName);
531      if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
532        _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
533      }
534      else {
535        _outFile->open(QIODevice::WriteOnly);
536      }
537      _out = new QTextStream(_outFile);
538      _out->setRealNumberNotation(QTextStream::FixedNotation);
539    }
540  }
541  else {
542    delete _out;     _out     = 0;
543    delete _outFile; _outFile = 0;
544  }
545}
546
547// Output into the Miscellaneous socket
548////////////////////////////////////////////////////////////////////////////
549void bncCaster::slotNewRawData(QByteArray staID, QByteArray data) {
550  if (_miscSockets && (_miscMount == "ALL" || _miscMount == staID)) {
551    QMutableListIterator<QTcpSocket*> is(*_miscSockets);
552    while (is.hasNext()) {
553      QTcpSocket* sock = is.next();
554      if (sock->state() == QAbstractSocket::ConnectedState) {
555        sock->write(data);
556      }
557      else if (sock->state() != QAbstractSocket::ConnectingState) {
558        delete sock;
559        is.remove();
560      }
561    }
562  }
563}
564
565// New Connection
566////////////////////////////////////////////////////////////////////////////
567void bncCaster::slotNewMiscConnection() {
568  _miscSockets->push_back( _miscServer->nextPendingConnection() );
569  emit( newMessage(QString("New client connection on Miscellaneous Output Port: # %1")
570                   .arg(_miscSockets->size()).toLatin1(), true) );
571}
Note: See TracBrowser for help on using the repository browser.