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

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