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

Last change on this file since 8916 was 8916, checked in by stuerze, 4 years 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.