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

Last change on this file since 1529 was 1529, checked in by weber, 15 years ago

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