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

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

minor changes

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