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

Last change on this file since 10911 was 10911, checked in by stuerze, 10 days ago

minor changes to solve an issue under windows operation system

File size: 18.5 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#ifndef WIN32
42#include <unistd.h>
43#else
44#define _HAS_STD_BYTE 0
45#include <windows.h>
46#endif
47#include <math.h>
48#include <iostream>
49#include <iomanip>
50#include <sstream>
51
52#include "bnccaster.h"
53#include "bncrinex.h"
54#include "bnccore.h"
55#include "bncgetthread.h"
56#include "bncutils.h"
57#include "bncsettings.h"
58
59using namespace std;
60
61// Constructor
62////////////////////////////////////////////////////////////////////////////
63bncCaster::bncCaster() {
64
65 bncSettings settings;
66
67 connect(this, SIGNAL(newMessage(QByteArray,bool)),
68 BNC_CORE, SLOT(slotMessage(const QByteArray,bool)));
69
70 _outFile = 0;
71 _out = 0;
72 reopenOutFile();
73
74 // FeedEngine output port
75 // ----------------------
76 int port = settings.value("outPort").toInt();
77 if (port != 0) {
78 _server = new QTcpServer;
79 _server->setProxy(QNetworkProxy::NoProxy);
80 if ( !_server->listen(QHostAddress::LocalHost, port) ) {
81 QString message = "bncCaster: Cannot listen on sync port "
82 + QByteArray::number(port) + ": "
83 + _server->errorString();
84 emit newMessage(message.toLatin1(), 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 _uServer->setProxy(QNetworkProxy::NoProxy);
98 if ( !_uServer->listen(QHostAddress::LocalHost, uPort) ) {
99 QString message = "bncCaster: Cannot listen on usync port "
100 + QByteArray::number(uPort) + ": "
101 + _uServer->errorString();
102 emit newMessage(message.toLatin1(), true);
103 }
104 connect(_uServer, SIGNAL(newConnection()), this, SLOT(slotNewUConnection()));
105 _uSockets = new QList<QTcpSocket*>;
106 }
107 else {
108 _uServer = 0;
109 _uSockets = 0;
110 }
111
112 _outLockTime = settings.value("outLockTime",false).toBool();
113 _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
114 _outWait = settings.value("outWait").toDouble();
115 if (_outWait <= 0.0) {
116 _outWait = 0.01;
117 }
118 _confInterval = -1;
119
120 // Miscellaneous output port
121 // -------------------------
122 _miscMount = settings.value("miscMount").toString();
123 _miscPort = settings.value("miscPort").toInt();
124 if (!_miscMount.isEmpty() && _miscPort != 0) {
125 _miscServer = new QTcpServer;
126 _miscServer->setProxy(QNetworkProxy::NoProxy);
127 if ( !_miscServer->listen(QHostAddress::LocalHost, _miscPort) ) {
128 QString message = "bncCaster: Cannot listen on miscellaneous output port "
129 + QByteArray::number(_miscPort) + ": "
130 + _miscServer->errorString();
131 emit newMessage(message.toLatin1(), true);
132 }
133 connect(_miscServer, SIGNAL(newConnection()), this, SLOT(slotNewMiscConnection()));
134 _miscSockets = new QList<QTcpSocket*>;
135 }
136 else {
137 _miscServer = 0;
138 _miscSockets = 0;
139 }
140
141 // RawUploadCaster
142 _bncRawUploadCaster = new bncRawUploadCaster();
143
144}
145
146// Destructor
147////////////////////////////////////////////////////////////////////////////
148bncCaster::~bncCaster() {
149
150 QMutexLocker locker(&_mutex);
151
152 QListIterator<bncGetThread*> it(_threads);
153
154 while(it.hasNext()){
155 bncGetThread* thread = it.next();
156 disconnect(thread, 0, 0, 0);
157 _staIDs.removeAll(thread->staID());
158 _threads.removeAll(thread);
159 thread->terminate();
160 }
161
162 delete _out;
163 delete _outFile;
164 delete _server;
165 delete _sockets;
166 delete _uServer;
167 delete _uSockets;
168 delete _miscServer;
169 delete _miscSockets;
170 delete _bncRawUploadCaster;
171}
172
173// New Observations
174////////////////////////////////////////////////////////////////////////////
175void bncCaster::slotNewObs(const QByteArray staID, QList<t_satObs> obsList) {
176
177 QMutexLocker locker(&_mutex);
178
179 reopenOutFile();
180
181 unsigned index = 0;
182 QMutableListIterator<t_satObs> it(obsList);
183 while (it.hasNext()) {
184 ++index;
185 t_satObs& obs = it.next();
186
187 // Rename the Station
188 // ------------------
189 obs._staID = staID.data();
190
191 // Update/Set Slip Counters
192 // ------------------------
193 setSlipCounters(obs);
194
195 // Output into the socket
196 // ----------------------
197 if (_uSockets) {
198
199 ostringstream oStr;
200 oStr.setf(ios::showpoint | ios::fixed);
201 oStr << obs._staID << " "
202 << setw(4) << obs._time.gpsw() << " "
203 << setw(14) << setprecision(7) << obs._time.gpssec() << " "
204 << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
205
206 string hlpStr = oStr.str();
207
208 QMutableListIterator<QTcpSocket*> is(*_uSockets);
209 while (is.hasNext()) {
210 QTcpSocket* sock = is.next();
211 if (sock->state() == QAbstractSocket::ConnectedState) {
212 int numBytes = hlpStr.length();
213 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
214 delete sock;
215 is.remove();
216 }
217 }
218 else if (sock->state() != QAbstractSocket::ConnectingState) {
219 delete sock;
220 is.remove();
221 }
222 }
223 }
224
225 // First time: set the _lastDumpTime
226 // ---------------------------------
227 if (!_lastDumpTime.valid()) {
228 _lastDumpTime = obs._time - 1.0;
229 }
230
231 // An old observation - throw it away
232 // ----------------------------------
233 if (obs._time <= _lastDumpTime) {
234 if (index == 1) {
235 bncSettings settings;
236 if ( !settings.value("outFile").toString().isEmpty() ||
237 !settings.value("outPort").toString().isEmpty() ) {
238 emit( newMessage(QString("%1: Old or erroneous observation epoch %2 thrown away from %3")
239 .arg(staID.data())
240 .arg(string(obs._time).c_str())
241 .arg(obs._prn.toString().c_str())
242 .toLatin1(), true) );
243 }
244 }
245 continue;
246 }
247
248 // Save the observation
249 // --------------------
250 _epochs[obs._time].append(obs);
251
252 // Dump Epochs
253 // -----------
254 if ((obs._time - _outWait) > _lastDumpTime) {
255 dumpEpochs(obs._time - _outWait);
256 _lastDumpTime = obs._time - _outWait;
257 }
258 }
259}
260
261// New Connection
262////////////////////////////////////////////////////////////////////////////
263void bncCaster::slotNewConnection() {
264 _sockets->push_back( _server->nextPendingConnection() );
265 emit( newMessage(QString("New client connection on sync port: # %1")
266 .arg(_sockets->size()).toLatin1(), true) );
267}
268
269void bncCaster::slotNewUConnection() {
270 _uSockets->push_back( _uServer->nextPendingConnection() );
271 emit( newMessage(QString("New client connection on usync port: # %1")
272 .arg(_uSockets->size()).toLatin1(), true) );
273}
274
275// Add New Thread
276////////////////////////////////////////////////////////////////////////////
277void bncCaster::addGetThread(bncGetThread* getThread, bool noNewThread) {
278
279 qRegisterMetaType<t_satObs>("t_satObs");
280 qRegisterMetaType< QList<t_satObs> >("QList<t_satObs>");
281
282 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
283 this, SLOT(slotNewObs(QByteArray, QList<t_satObs>)));
284
285 connect(getThread, SIGNAL(newObs(QByteArray, QList<t_satObs>)),
286 this, SIGNAL(newObs(QByteArray, QList<t_satObs>)));
287
288 connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
289 this, SLOT(slotNewRawData(QByteArray, QByteArray)));
290
291 connect(getThread, SIGNAL(newRawData(QByteArray, QByteArray)),
292 _bncRawUploadCaster, SLOT(slotNewRawData(QByteArray, QByteArray)));
293
294 connect(getThread, SIGNAL(getThreadFinished(QByteArray)),
295 this, SLOT(slotGetThreadFinished(QByteArray)));
296
297 _staIDs.push_back(getThread->staID());
298 _threads.push_back(getThread);
299
300 if (noNewThread) {
301 getThread->run();
302 }
303 else {
304 getThread->start();
305 }
306}
307
308// Get Thread destroyed
309////////////////////////////////////////////////////////////////////////////
310void bncCaster::slotGetThreadFinished(QByteArray staID) {
311 //QMutexLocker locker(&_mutex);
312
313 QListIterator<bncGetThread*> it(_threads);
314 while (it.hasNext()) {
315 bncGetThread* thread = it.next();
316 if (thread->staID() == staID) {
317 _threads.removeOne(thread);
318 }
319 }
320 _staIDs.removeAll(staID);
321 emit( newMessage(
322 QString("Decoding %1 stream(s)").arg(_staIDs.size()).toLatin1(), true) );
323 if (_staIDs.size() == 0) {
324 emit getThreadsFinished();
325 }
326
327}
328
329// Dump Complete Epochs
330////////////////////////////////////////////////////////////////////////////
331void bncCaster::dumpEpochs(const bncTime& maxTime) {
332
333 QMutableMapIterator<bncTime, QList<t_satObs> > itEpo(_epochs);
334 while (itEpo.hasNext()) {
335 itEpo.next();
336 const bncTime& epoTime = itEpo.key();
337 if (epoTime <= maxTime) {
338 const QList<t_satObs>& allObs = itEpo.value();
339 int sec = int(nint(epoTime.gpssec()*10));
340 if ( (_out || _sockets) && (sec % (_samplingRateMult10) == 0) ) {
341 QListIterator<t_satObs> it(allObs);
342 bool firstObs = true;
343 while (it.hasNext()) {
344 const t_satObs& obs = it.next();
345
346 ostringstream oStr;
347 oStr.setf(ios::showpoint | ios::fixed);
348 if (firstObs) {
349 firstObs = false;
350 oStr << "> " << obs._time.gpsw() << ' '
351 << setprecision(7) << obs._time.gpssec() << endl;
352 }
353 oStr << obs._staID << ' '
354 << bncRinex::asciiSatLine(obs,_outLockTime) << endl;
355 if (!it.hasNext()) {
356 oStr << endl;
357 }
358 string hlpStr = oStr.str();
359
360 // Output into the File
361 // --------------------
362 if (_out) {
363 *_out << hlpStr.c_str();
364 _out->flush();
365 }
366
367 // Output into the socket
368 // ----------------------
369 if (_sockets) {
370 QMutableListIterator<QTcpSocket*> is(*_sockets);
371 while (is.hasNext()) {
372 QTcpSocket* sock = is.next();
373 if (sock->state() == QAbstractSocket::ConnectedState) {
374 int numBytes = hlpStr.length();
375 if (myWrite(sock, hlpStr.c_str(), numBytes) != numBytes) {
376 delete sock;
377 is.remove();
378 }
379 }
380 else if (sock->state() != QAbstractSocket::ConnectingState) {
381 delete sock;
382 is.remove();
383 }
384 }
385 }
386 }
387 }
388 //_epochs.remove(epoTime);
389 itEpo.remove();
390 }
391 }
392}
393
394// Reread configuration (private slot)
395////////////////////////////////////////////////////////////////////////////
396void bncCaster::slotReadMountPoints() {
397
398 bncSettings settings;
399 settings.reRead();
400
401 readMountPoints();
402}
403
404// Read Mountpoints
405////////////////////////////////////////////////////////////////////////////
406void bncCaster::readMountPoints() {
407
408 bncSettings settings;
409
410 // Reread several options
411 // ----------------------
412 _samplingRateMult10 = int(settings.value("outSampl").toString().split("sec").first().toDouble() * 10.0);
413
414 _outWait = settings.value("outWait").toDouble();
415 if (_outWait <= 0.0) {
416 _outWait = 0.01;
417 }
418
419 // Add new mountpoints
420 // -------------------
421 int iMount = -1;
422 QListIterator<QString> it(settings.value("mountPoints").toStringList());
423 while (it.hasNext()) {
424 ++iMount;
425 QStringList hlp = it.next().split(" ");
426 if (hlp.size() < 7) continue;
427 QUrl url(hlp[0]);
428
429 // Does it already exist?
430 // ----------------------
431 bool existFlg = false;
432 QListIterator<bncGetThread*> iTh(_threads);
433 while (iTh.hasNext()) {
434 bncGetThread* thread = iTh.next();
435 if (thread->mountPoint() == url) {
436 existFlg = true;
437 break;
438 }
439 }
440
441 // New bncGetThread
442 // ----------------
443 if (!existFlg) {
444 QByteArray format = hlp[1].toLatin1();
445 QByteArray latitude = hlp[3].toLatin1();
446 QByteArray longitude = hlp[4].toLatin1();
447 QByteArray nmea = hlp[5].toLatin1();
448 QByteArray ntripVersion = hlp[6].toLatin1();
449
450 bncGetThread* getThread = new bncGetThread(url, format, latitude,
451 longitude, nmea, ntripVersion);
452 addGetThread(getThread);
453
454 }
455 }
456
457 // Remove mountpoints
458 // ------------------
459 QListIterator<bncGetThread*> iTh(_threads);
460 while (iTh.hasNext()) {
461 bncGetThread* thread = iTh.next();
462
463 bool existFlg = false;
464 QListIterator<QString> it(settings.value("mountPoints").toStringList());
465 while (it.hasNext()) {
466 QStringList hlp = it.next().split(" ");
467 if (hlp.size() <= 1) continue;
468 QUrl url(hlp[0]);
469
470 if (thread->mountPoint() == url) {
471 existFlg = true;
472 break;
473 }
474 }
475
476 if (!existFlg) {
477 disconnect(thread, 0, 0, 0);
478 _staIDs.removeAll(thread->staID());
479 _threads.removeAll(thread);
480 thread->terminate();
481 }
482 }
483
484 emit mountPointsRead(_threads);
485 emit(newMessage(QString("Configuration read: " + BNC_CORE->confFileName() + ", %1 stream(s)").arg(_threads.count()).toLatin1(), true));
486
487 // (Re-) Start the configuration timer
488 // -----------------------------------
489 if (settings.value("onTheFlyInterval").toString() == "no") {
490 return;
491 }
492
493 int ms = 0;
494 if (_confInterval != -1) {
495 ms = 1000 * _confInterval;
496 }
497 else {
498 QTime currTime = currentDateAndTimeGPS().time();
499 QTime nextShotTime;
500 if (settings.value("onTheFlyInterval").toString() == "1 min") {
501 _confInterval = 60;
502 nextShotTime = QTime(currTime.hour(), currTime.minute()+1, 0);
503 }
504 else if (settings.value("onTheFlyInterval").toString() == "5 min") {
505 _confInterval = 300;
506 nextShotTime = QTime(currTime.hour(), currTime.minute()+5, 0);
507 }
508 else if (settings.value("onTheFlyInterval").toString() == "1 hour") {
509 _confInterval = 3600;
510 nextShotTime = QTime(currTime.hour()+1, 0, 0);
511 }
512 else {
513 _confInterval = 86400;
514 nextShotTime = QTime(23, 59, 59, 999);
515 }
516
517 ms = currTime.msecsTo(nextShotTime);
518 if (ms < 30000) {
519 ms = 30000;
520 }
521 }
522
523 QTimer::singleShot(ms, this, SLOT(slotReadMountPoints()));
524}
525
526//
527////////////////////////////////////////////////////////////////////////////
528int bncCaster::myWrite(QTcpSocket* sock, const char* buf, int bufLen) {
529 sock->write(buf, bufLen);
530 for (int ii = 1; ii <= 10; ii++) {
531 if (sock->waitForBytesWritten(10)) { // wait 10 ms
532 return bufLen;
533 }
534 }
535 return -1;
536}
537
538//
539////////////////////////////////////////////////////////////////////////////
540void bncCaster::reopenOutFile() {
541
542 bncSettings settings;
543
544 QString outFileName = settings.value("outFile").toString();
545 if ( !outFileName.isEmpty() ) {
546 expandEnvVar(outFileName);
547 if (!_outFile || _outFile->fileName() != outFileName) {
548 delete _out;
549 delete _outFile;
550 _outFile = new QFile(outFileName);
551 if ( Qt::CheckState(settings.value("rnxAppend").toInt()) == Qt::Checked) {
552 _outFile->open(QIODevice::WriteOnly | QIODevice::Append);
553 }
554 else {
555 _outFile->open(QIODevice::WriteOnly);
556 }
557 _out = new QTextStream(_outFile);
558 _out->setRealNumberNotation(QTextStream::FixedNotation);
559 }
560 }
561 else {
562 delete _out; _out = 0;
563 delete _outFile; _outFile = 0;
564 }
565}
566
567// Output into the Miscellaneous socket
568////////////////////////////////////////////////////////////////////////////
569void bncCaster::slotNewRawData(QByteArray staID, QByteArray data) {
570 if (_miscSockets && (_miscMount == "ALL" || _miscMount == staID)) {
571 QMutableListIterator<QTcpSocket*> is(*_miscSockets);
572 while (is.hasNext()) {
573 QTcpSocket* sock = is.next();
574 if (sock->state() == QAbstractSocket::ConnectedState) {
575 sock->write(data);
576 }
577 else if (sock->state() != QAbstractSocket::ConnectingState) {
578 delete sock;
579 is.remove();
580 }
581 }
582 }
583}
584
585// New Connection
586////////////////////////////////////////////////////////////////////////////
587void bncCaster::slotNewMiscConnection() {
588 _miscSockets->push_back( _miscServer->nextPendingConnection() );
589 emit( newMessage(QString("New client connection on Miscellaneous Output Port: # %1")
590 .arg(_miscSockets->size()).toLatin1(), true) );
591}
592
593// Set/Update Slip Counters
594////////////////////////////////////////////////////////////////////////////
595void bncCaster::setSlipCounters(t_satObs& obs) {
596
597 double minLockTime = -1.0;
598 for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
599 const t_frqObs* frqObs = obs._obs[ii];
600 if (frqObs->_lockTimeValid) {
601 if (minLockTime == -1.0 || minLockTime < frqObs->_lockTime) {
602 minLockTime = frqObs->_lockTime;
603 }
604 }
605 }
606
607 if (minLockTime == -1.0) {
608 return;
609 }
610
611 QMap<t_prn, double>& ltMap = _lockTimeMap[obs._staID];
612 QMap<t_prn, int>& jcMap = _jumpCounterMap[obs._staID];
613 QMap<t_prn, double>::const_iterator it = ltMap.find(obs._prn);
614 if (it == ltMap.end()) { // init satellite
615 ltMap[obs._prn] = minLockTime;
616 jcMap[obs._prn] = 0;
617 }
618 else {
619 if (minLockTime < ltMap[obs._prn]) {
620 jcMap[obs._prn] += 1;
621 if (jcMap[obs._prn] > 9999) {
622 jcMap[obs._prn] = 0;
623 }
624 }
625 ltMap[obs._prn] = minLockTime;
626 }
627 for (unsigned ii = 0; ii < obs._obs.size(); ii++) {
628 t_frqObs* frqObs = obs._obs[ii];
629 frqObs->_slipCounter = jcMap[obs._prn];
630 }
631}
Note: See TracBrowser for help on using the repository browser.