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

Last change on this file since 7933 was 7854, checked in by stuerze, 9 years ago

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