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

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