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

Last change on this file since 9124 was 9124, checked in by stuerze, 7 months ago

minor changes regarding latency check, ssr and long mountpoint names

File size: 18.2 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    _server->setProxy(QNetworkProxy::NoProxy);
74    if ( !_server->listen(QHostAddress::Any, port) ) {
75      QString message = "bncCaster: Cannot listen on sync port "
76                      + QByteArray::number(port) + ": "
77                      + _server->errorString();
78      emit newMessage(message.toLatin1(), true);
79    }
80    connect(_server, SIGNAL(newConnection()), this, SLOT(slotNewConnection()));
81    _sockets = new QList<QTcpSocket*>;
82  }
83  else {
84    _server  = 0;
85    _sockets = 0;
86  }
87
88  int uPort = settings.value("outUPort").toInt();
89  if (uPort != 0) {
90    _uServer = new QTcpServer;
91    _uServer->setProxy(QNetworkProxy::NoProxy);
92    if ( !_uServer->listen(QHostAddress::Any, uPort) ) {
93      QString message = "bncCaster: Cannot listen on usync port "
94                      + QByteArray::number(uPort) + ": "
95                      + _uServer->errorString();
96      emit newMessage(message.toLatin1(), true);
97    }
98    connect(_uServer, SIGNAL(newConnection()), this, SLOT(slotNewUConnection()));
99    _uSockets = new QList<QTcpSocket*>;
100  }
101  else {
102    _uServer  = 0;
103    _uSockets = 0;
104  }
105
106  _outLockTime = settings.value("outLockTime",false).toBool();
107  _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
108  _outWait      = settings.value("outWait").toDouble();
109  if (_outWait <= 0.0) {
110    _outWait = 0.01;
111  }
112  _confInterval = -1;
113
114  // Miscellaneous output port
115  // -------------------------
116  _miscMount = settings.value("miscMount").toString();
117  _miscPort  = settings.value("miscPort").toInt();
118  if (!_miscMount.isEmpty() && _miscPort != 0) {
119    _miscServer = new QTcpServer;
120    _miscServer->setProxy(QNetworkProxy::NoProxy);
121    if ( !_miscServer->listen(QHostAddress::Any, _miscPort) ) {
122      QString message = "bncCaster: Cannot listen on miscellaneous output port "
123                      + QByteArray::number(_miscPort) + ": "
124                      + _miscServer->errorString();
125      emit newMessage(message.toLatin1(), true);
126    }
127    connect(_miscServer, SIGNAL(newConnection()), this, SLOT(slotNewMiscConnection()));
128    _miscSockets = new QList<QTcpSocket*>;
129  }
130  else {
131    _miscServer  = 0;
132    _miscSockets = 0;
133  }
134}
135
136// Destructor
137////////////////////////////////////////////////////////////////////////////
138bncCaster::~bncCaster() {
139
140  QMutexLocker locker(&_mutex);
141
142  QListIterator<bncGetThread*> it(_threads);
143  while(it.hasNext()){
144    bncGetThread* thread = it.next();
145    disconnect(thread, 0, 0, 0);
146    _staIDs.removeAll(thread->staID());
147    _threads.removeAll(thread);
148    thread->terminate();
149  }
150  delete _out;
151  delete _outFile;
152  delete _server;
153  delete _sockets;
154  delete _uServer;
155  delete _uSockets;
156  delete _miscServer;
157  delete _miscSockets;
158}
159
160// New Observations
161////////////////////////////////////////////////////////////////////////////
162void bncCaster::slotNewObs(const QByteArray staID, QList<t_satObs> obsList) {
163
164  QMutexLocker locker(&_mutex);
165
166  reopenOutFile();
167
168  unsigned index = 0;
169  QMutableListIterator<t_satObs> it(obsList);
170  while (it.hasNext()) {
171    ++index;
172    t_satObs& obs = it.next();
173
174    // Rename the Station
175    // ------------------
176    obs._staID = staID.data();
177
178    // Update/Set Slip Counters
179    // ------------------------
180    setSlipCounters(obs);
181
182    // Output into the socket
183    // ----------------------
184    if (_uSockets) {
185
186      ostringstream oStr;
187      oStr.setf(ios::showpoint | ios::fixed);
188      oStr << obs._staID                                        << " "
189           << setw(4)  << obs._time.gpsw()                      << " "
190           << setw(14) << setprecision(7) << obs._time.gpssec() << " "
191           << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
192
193      string hlpStr = oStr.str();
194
195      QMutableListIterator<QTcpSocket*> is(*_uSockets);
196      while (is.hasNext()) {
197        QTcpSocket* sock = is.next();
198        if (sock->state() == QAbstractSocket::ConnectedState) {
199          int numBytes = hlpStr.length();
200          if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
201            delete sock;
202            is.remove();
203          }
204        }
205        else if (sock->state() != QAbstractSocket::ConnectingState) {
206          delete sock;
207          is.remove();
208        }
209      }
210    }
211
212    // First time: set the _lastDumpTime
213    // ---------------------------------
214    if (!_lastDumpTime.valid()) {
215      _lastDumpTime = obs._time - 1.0;
216    }
217
218    // An old observation - throw it away
219    // ----------------------------------
220    if (obs._time <= _lastDumpTime) {
221      if (index == 1) {
222        bncSettings settings;
223        if ( !settings.value("outFile").toString().isEmpty() ||
224             !settings.value("outPort").toString().isEmpty() ) {
225          emit( newMessage(QString("%1: Old or erroneous observation epoch %2 thrown away from %3")
226                           .arg(staID.data())
227                           .arg(string(obs._time).c_str())
228                           .arg(obs._prn.toString().c_str())
229                             .toLatin1(), true) );
230        }
231      }
232      continue;
233    }
234
235    // Save the observation
236    // --------------------
237    _epochs[obs._time].append(obs);
238
239    // Dump Epochs
240    // -----------
241    if ((obs._time - _outWait) > _lastDumpTime) {
242      dumpEpochs(obs._time - _outWait);
243      _lastDumpTime = obs._time - _outWait;
244    }
245  }
246}
247
248// New Connection
249////////////////////////////////////////////////////////////////////////////
250void bncCaster::slotNewConnection() {
251  _sockets->push_back( _server->nextPendingConnection() );
252  emit( newMessage(QString("New client connection on sync port: # %1")
253                   .arg(_sockets->size()).toLatin1(), true) );
254}
255
256void bncCaster::slotNewUConnection() {
257  _uSockets->push_back( _uServer->nextPendingConnection() );
258  emit( newMessage(QString("New client connection on usync port: # %1")
259                   .arg(_uSockets->size()).toLatin1(), true) );
260}
261
262// Add New Thread
263////////////////////////////////////////////////////////////////////////////
264void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
265
266  qRegisterMetaType<t_satObs>("t_satObs");
267  qRegisterMetaType< QList<t_satObs> >("QList<t_satObs>");
268
269  connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
270          this,      SLOT(slotNewObs(QByteArray, QList<t_satObs>)));
271
272  connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
273          this,      SIGNAL(newObs(QByteArray, QList<t_satObs>)));
274
275  connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
276          this,      SLOT(slotNewRawData(QByteArray, QByteArray)));
277
278  connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
279          this, SLOT(slotGetThreadFinished(QByteArray)));
280
281  _staIDs.push_back(getThread->staID());
282  _threads.push_back(getThread);
283
284  if (noNewThread) {
285    getThread->run();
286  }
287  else {
288    getThread->start();
289  }
290}
291
292// Get Thread destroyed
293////////////////////////////////////////////////////////////////////////////
294void bncCaster::slotGetThreadFinished(QByteArray staID) {
295  QMutexLocker locker(&_mutex);
296
297  QListIterator<bncGetThread*> it(_threads);
298  while (it.hasNext()) {
299    bncGetThread* thread = it.next();
300    if (thread->staID() == staID) {
301      _threads.removeOne(thread);
302    }
303  }
304
305  _staIDs.removeAll(staID);
306  emit( newMessage(
307           QString("Decoding %1 stream(s)").arg(_staIDs.size()).toLatin1(), true) );
308  if (_staIDs.size() == 0) {
309    emit(newMessage("bncCaster: Last get thread terminated", true));
310    emit getThreadsFinished();
311  }
312}
313
314// Dump Complete Epochs
315////////////////////////////////////////////////////////////////////////////
316void bncCaster::dumpEpochs(const bncTime& maxTime) {
317
318  QMutableMapIterator<bncTime, QList<t_satObs> > itEpo(_epochs);
319  while (itEpo.hasNext()) {
320    itEpo.next();
321    const bncTime& epoTime = itEpo.key();
322    if (epoTime <= maxTime) {
323      const QList<t_satObs>& allObs = itEpo.value();
324      int sec = int(nint(epoTime.gpssec()*10));
325      if ( (_out || _sockets) && (sec % (_samplingRateMult10) == 0) ) {
326        QListIterator<t_satObs> it(allObs);
327        bool firstObs = true;
328        while (it.hasNext()) {
329          const t_satObs& obs = it.next();
330
331          ostringstream oStr;
332          oStr.setf(ios::showpoint | ios::fixed);
333          if (firstObs) {
334            firstObs = false;
335            oStr << "> " << obs._time.gpsw() << ' '
336                 << setprecision(7) << obs._time.gpssec() << endl;
337          }
338          oStr << obs._staID << ' '
339               << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
340          if (!it.hasNext()) {
341            oStr << endl;
342          }
343          string hlpStr = oStr.str();
344
345          // Output into the File
346          // --------------------
347          if (_out) {
348            *_out << hlpStr.c_str();
349            _out->flush();
350          }
351
352          // Output into the socket
353          // ----------------------
354          if (_sockets) {
355            QMutableListIterator<QTcpSocket*> is(*_sockets);
356            while (is.hasNext()) {
357              QTcpSocket* sock = is.next();
358              if (sock->state() == QAbstractSocket::ConnectedState) {
359                int numBytes = hlpStr.length();
360                if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
361                  delete sock;
362                  is.remove();
363                }
364              }
365              else if (sock->state() != QAbstractSocket::ConnectingState) {
366                delete sock;
367                is.remove();
368              }
369            }
370          }
371        }
372      }
373      _epochs.remove(epoTime);
374    }
375  }
376}
377
378// Reread configuration (private slot)
379////////////////////////////////////////////////////////////////////////////
380void bncCaster::slotReadMountPoints() {
381
382  bncSettings settings;
383  settings.reRead();
384
385  readMountPoints();
386}
387
388// Read Mountpoints
389////////////////////////////////////////////////////////////////////////////
390void bncCaster::readMountPoints() {
391
392  bncSettings settings;
393
394  // Reread several options
395  // ----------------------
396  _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
397  _outWait      = settings.value("outWait").toInt();
398  if (_outWait < 1) {
399    _outWait = 1;
400  }
401
402  // Add new mountpoints
403  // -------------------
404  int iMount = -1;
405  QListIterator<QString> it(settings.value("mountPoints").toStringList());
406  while (it.hasNext()) {
407    ++iMount;
408    QStringList hlp = it.next().split(" ");
409    if (hlp.size() < 7) continue;
410    QUrl url(hlp[0]);
411
412    // Does it already exist?
413    // ----------------------
414    bool existFlg = false;
415    QListIterator<bncGetThread*> iTh(_threads);
416    while (iTh.hasNext()) {
417      bncGetThread* thread = iTh.next();
418      if (thread->mountPoint() == url) {
419        existFlg = true;
420        break;
421      }
422    }
423
424    // New bncGetThread
425    // ----------------
426    if (!existFlg) {
427      QByteArray format    = hlp[1].toLatin1();
428      QByteArray latitude  = hlp[3].toLatin1();
429      QByteArray longitude = hlp[4].toLatin1();
430      QByteArray nmea      = hlp[5].toLatin1();
431      QByteArray ntripVersion = hlp[6].toLatin1();
432
433      bncGetThread* getThread = new bncGetThread(url, format, latitude,
434                                        longitude, nmea, ntripVersion);
435      addGetThread(getThread);
436
437    }
438  }
439
440  // Remove mountpoints
441  // ------------------
442  QListIterator<bncGetThread*> iTh(_threads);
443  while (iTh.hasNext()) {
444    bncGetThread* thread = iTh.next();
445
446    bool existFlg = false;
447    QListIterator<QString> it(settings.value("mountPoints").toStringList());
448    while (it.hasNext()) {
449      QStringList hlp = it.next().split(" ");
450      if (hlp.size() <= 1) continue;
451      QUrl url(hlp[0]);
452
453      if (thread->mountPoint() == url) {
454        existFlg = true;
455        break;
456      }
457    }
458
459    if (!existFlg) {
460      disconnect(thread, 0, 0, 0);
461      _staIDs.removeAll(thread->staID());
462      _threads.removeAll(thread);
463      thread->terminate();
464    }
465  }
466
467  emit mountPointsRead(_threads);
468  emit( newMessage(QString("Configuration read: "
469                           + BNC_CORE->confFileName()
470                           + ", %1 stream(s)")
471                            .arg(_threads.count()).toLatin1(), true) );
472
473  // (Re-) Start the configuration timer
474  // -----------------------------------
475  if      (settings.value("onTheFlyInterval").toString() == "no") {
476    return;
477  }
478
479  int ms = 0;
480  if (_confInterval != -1) {
481    ms = 1000 * _confInterval;
482  }
483  else {
484    QTime currTime = currentDateAndTimeGPS().time();
485    QTime nextShotTime;
486    if      (settings.value("onTheFlyInterval").toString() == "1 min") {
487      _confInterval = 60;
488      nextShotTime = QTime(currTime.hour(), currTime.minute()+1, 0);
489    }
490    else if (settings.value("onTheFlyInterval").toString() == "5 min") {
491      _confInterval = 300;
492      nextShotTime = QTime(currTime.hour(), currTime.minute()+5, 0);
493    }
494    else if (settings.value("onTheFlyInterval").toString() == "1 hour") {
495      _confInterval = 3600;
496      nextShotTime = QTime(currTime.hour()+1, 0, 0);
497    }
498    else {
499      _confInterval = 86400;
500      nextShotTime = QTime(23, 59, 59, 999);
501    }
502
503    ms = currTime.msecsTo(nextShotTime);
504    if (ms < 30000) {
505      ms = 30000;
506    }
507  }
508
509  QTimer::singleShot(ms, this, SLOT(slotReadMountPoints()));
510}
511
512//
513////////////////////////////////////////////////////////////////////////////
514int bncCaster::myWrite(QTcpSocket* sock, const char* buf, int bufLen) {
515  sock->write(buf, bufLen);
516  for (int ii = 1; ii <= 10; ii++) {
517    if (sock->waitForBytesWritten(10)) {  // wait 10 ms
518      return bufLen;
519    }
520  }
521  return -1;
522}
523
524//
525////////////////////////////////////////////////////////////////////////////
526void bncCaster::reopenOutFile() {
527
528  bncSettings settings;
529
530  QString outFileName = settings.value("outFile").toString();
531  if ( !outFileName.isEmpty() ) {
532    expandEnvVar(outFileName);
533    if (!_outFile || _outFile->fileName() != outFileName) {
534      delete _out;
535      delete _outFile;
536      _outFile = new QFile(outFileName);
537      if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
538        _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
539      }
540      else {
541        _outFile->open(QIODevice::WriteOnly);
542      }
543      _out = new QTextStream(_outFile);
544      _out->setRealNumberNotation(QTextStream::FixedNotation);
545    }
546  }
547  else {
548    delete _out;     _out     = 0;
549    delete _outFile; _outFile = 0;
550  }
551}
552
553// Output into the Miscellaneous socket
554////////////////////////////////////////////////////////////////////////////
555void bncCaster::slotNewRawData(QByteArray staID, QByteArray data) {
556  if (_miscSockets && (_miscMount == "ALL" || _miscMount == staID)) {
557    QMutableListIterator<QTcpSocket*> is(*_miscSockets);
558    while (is.hasNext()) {
559      QTcpSocket* sock = is.next();
560      if (sock->state() == QAbstractSocket::ConnectedState) {
561        sock->write(data);
562      }
563      else if (sock->state() != QAbstractSocket::ConnectingState) {
564        delete sock;
565        is.remove();
566      }
567    }
568  }
569}
570
571// New Connection
572////////////////////////////////////////////////////////////////////////////
573void bncCaster::slotNewMiscConnection() {
574  _miscSockets->push_back( _miscServer->nextPendingConnection() );
575  emit( newMessage(QString("New client connection on Miscellaneous Output Port: # %1")
576                   .arg(_miscSockets->size()).toLatin1(), true) );
577}
578
579// Set/Update Slip Counters
580////////////////////////////////////////////////////////////////////////////
581void bncCaster::setSlipCounters(t_satObs& obs) {
582
583  double minLockTime = -1.0;
584  for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
585    const t_frqObs* frqObs = obs._obs[ii];
586    if (frqObs->_lockTimeValid) {
587      if (minLockTime == -1.0 || minLockTime < frqObs->_lockTime) {
588        minLockTime = frqObs->_lockTime;
589      }
590    }
591  }
592
593  if (minLockTime == -1.0) {
594    return;
595  }
596
597  QMap<t_prn, double>& ltMap = _lockTimeMap[obs._staID];
598  QMap<t_prn, int>&    jcMap = _jumpCounterMap[obs._staID];
599  QMap<t_prn, double>::const_iterator it = ltMap.find(obs._prn);
600  if (it == ltMap.end()) { // init satellite
601    ltMap[obs._prn] = minLockTime;
602    jcMap[obs._prn] = 0;
603  }
604  else {
605    if (minLockTime < ltMap[obs._prn]) {
606      jcMap[obs._prn] += 1;
607      if (jcMap[obs._prn] > 9999) {
608        jcMap[obs._prn] = 0;
609      }
610    }
611    ltMap[obs._prn] = minLockTime;
612  }
613  for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
614    t_frqObs* frqObs = obs._obs[ii];
615    frqObs->_slipCounter = jcMap[obs._prn];
616  }
617}
Note: See TracBrowser for help on using the repository browser.