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

Last change on this file since 1390 was 1390, checked in by mervart, 15 years ago

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