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

Last change on this file since 4720 was 4536, checked in by weber, 12 years ago

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