source: ntrip/trunk/BNC/bnccaster.cpp@ 2829

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