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

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