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

Last change on this file since 8917 was 8917, checked in by stuerze, 4 years ago

minor changes

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