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

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