gempa_public/libcapsclient
7
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
44609d367f6a6de37cb92a1d3565b0c0cccbb36f
libcapsclient/libs/gempa/caps/plugin.cpp
Enrico Ellguth 44609d367f Update to version 3.2
2026-03-18 15:18:05 +01:00

1698 lines
46 KiB
C++
Raw Blame History

/***************************************************************************
* Copyright (C) 2013 by gempa GmbH *
* *
* All Rights Reserved. *
* *
* NOTICE: All information contained herein is, and remains *
* the property of gempa GmbH and its suppliers, if any. The intellectual *
* and technical concepts contained herein are proprietary to gempa GmbH *
* and its suppliers. *
* Dissemination of this information or reproduction of this material *
* is strictly forbidden unless prior written permission is obtained *
* from gempa GmbH. *
***************************************************************************/
#include <gempa/caps/endianess.h>
#include <gempa/caps/log.h>
#include <gempa/caps/plugin.h>
#include <gempa/caps/utils.h>
#ifdef WIN32
#define EWOULDBLOCK WSAEWOULDBLOCK
#endif
#if !defined(CAPS_FEATURES_ANY) || CAPS_FEATURES_ANY
#include <gempa/caps/anypacket.h>
#endif
#if !defined(CAPS_FEATURES_MSEED) || CAPS_FEATURES_MSEED
#include <gempa/caps/mseedpacket.h>
#endif
#if !defined(CAPS_FEATURES_RAW) || CAPS_FEATURES_RAW
#include <gempa/caps/rawpacket.h>
#endif
#if !defined(CAPS_FEATURES_RTCM2) || CAPS_FEATURES_RTCM2
#include <gempa/caps/rtcm2packet.h>
#endif
#if !defined(CAPS_SC_LOGGING) || CAPS_SC_LOGGING
#include <seiscomp/logging/log.h>
#endif
#include <boost/version.hpp>
#include <boost/algorithm/string.hpp>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <cerrno>
#include <cstdarg>
using namespace std;
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
namespace fs = std::filesystem;
#endif
namespace {
#if !defined(CAPS_SC_LOGGING) || CAPS_SC_LOGGING
#define LOG_CHANNEL(out, fmt) \
va_list ap;\
va_start(ap, fmt);\
out(fmt, ap);\
va_end(ap)
#else
#define LOG_CHANNEL(out, fmt) \
va_list ap;\
va_start(ap, fmt);\
fprintf(stderr, #out" "); vfprintf(stderr, fmt, ap); fprintf(stderr, "\n");\
va_end(ap)
#endif
#if !defined(CAPS_SC_LOGGING) || CAPS_SC_LOGGING
void LogError(const char *fmt, ...) {
LOG_CHANNEL(SEISCOMP_VERROR, fmt);
}
void LogWarning(const char *fmt, ...) {
LOG_CHANNEL(SEISCOMP_VWARNING, fmt);
}
void LogNotice(const char *fmt, ...) {
LOG_CHANNEL(SEISCOMP_VNOTICE, fmt);
}
void LogInfo(const char *fmt, ...) {
LOG_CHANNEL(SEISCOMP_VINFO, fmt);
}
void LogDebug(const char *fmt, ...) {
LOG_CHANNEL(SEISCOMP_VDEBUG, fmt);
}
#else
#define LOG_CHANNEL(out, fmt) \
va_list ap;\
va_start(ap, fmt);\
fprintf(stderr, #out" "); vfprintf(stderr, fmt, ap); fprintf(stderr, "\n");\
va_end(ap)
void LogError(const char *fmt, ...) {
LOG_CHANNEL(ERROR, fmt);
}
void LogWarning(const char *fmt, ...) {
LOG_CHANNEL(WARNING, fmt);
}
void LogNotice(const char *fmt, ...) {
LOG_CHANNEL(NOTICE, fmt);
}
void LogInfo(const char *fmt, ...) {
LOG_CHANNEL(INFO, fmt);
}
void LogDebug(const char *fmt, ...) {
LOG_CHANNEL(DEBUG, fmt);
}
#endif
/**
* @brief Validate stream ID components regarding the following
* pattern [a-z][A-Z][0-9][-,_]
* @param net The network code
* @param sta The station code
* @param loc The location code
* @param cha The channel code
* @return True on success
*/
bool validateStreamID(const std::string &net, const std::string &sta,
const std::string &loc, const std::string &cha) {
std::vector<const std::string*> SID = {&net, &sta, &loc, &cha};
for ( int i = 0; i < 4; ++i ) {
for ( const char *ch = SID[i]->c_str(); *ch != 0; ++ch ) {
if ( (*ch < 65 || *ch > 90) && // A-Z
(*ch < 48 || *ch > 57) && // 0-9
(*ch < 97 || *ch > 122) && // a-z
*ch != 45 && *ch != 95 ) { // -, _
CAPS_ERROR("%s.%s.%s.%s: Invalid character in token #%i (%s) of "
"stream id", net.c_str(), sta.c_str(), loc.c_str(),
cha.c_str(), i, SID[i]->c_str());
return false;
}
}
}
return true;
}
template <typename T>
inline std::string toString(const T &v) {
std::ostringstream os;
os << v;
return os.str();
}
}
namespace Gempa {
namespace CAPS {
namespace {
#define HELLO_REQUEST "HELLO\n"
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
bool createPath(const string &dir) {
try {
fs::path path(dir);
if ( fs::is_directory(path) ) {
return true;
}
return fs::create_directories(path);
} catch ( ... ) {
return false;
}
}
#endif
#if !defined(CAPS_FEATURES_BACKFILLING) || CAPS_FEATURES_BACKFILLING
//void dump(const Plugin::BackfillingBuffer &buf) {
// Plugin::BackfillingBuffer::const_iterator it;
// int idx = 0;
// for ( it = buf.begin(); it != buf.end(); ++it, ++idx ) {
// std::cerr << idx << ": " << (*it)->record->startTime().iso() << std::endl;
// }
//}
template<typename T> void dumpSamples(RawDataRecord *rec, size_t sampleCount) {
const vector<T> *data = reinterpret_cast<vector<T>* >(rec->data());
for ( size_t i = 0; i < sampleCount; ++i ) {
cout << data->at(i) << endl;
}
}
void dumpSList(Packet *packet) {
RawDataRecord *rec = static_cast<RawDataRecord*>(packet->record.get());
string id = packet->streamID;
boost::replace_all(id, ".", "_");
string dataType = packet->record->formatName();
boost::replace_all(dataType, "RAW/", "");
DataType dt = rec->header()->dataType;
size_t sampleCount = rec->data()->size() / dataTypeSize(dt);
cout << "TIMESERIES " << id << "_R, "
<< sampleCount << " samples, "
<< static_cast<float>(rec->header()->samplingFrequencyNumerator) / rec->header()->samplingFrequencyDenominator << " sps, "
<< packet->record->startTime().iso() << " SLIST, "
<< dataType << ", " << packet->uom << endl;
if ( dt == DT_INT8 ) {
dumpSamples<int8_t>(rec, sampleCount);
}
else if ( dt == DT_INT16 ) {
dumpSamples<int16_t>(rec, sampleCount);
}
else if ( dt == DT_INT32 ) {
dumpSamples<int32_t>(rec, sampleCount);
}
else if ( dt == DT_INT64 ) {
dumpSamples<int64_t>(rec, sampleCount);
}
else if ( dt == DT_FLOAT ) {
dumpSamples<float>(rec, sampleCount);
}
else if ( dt == DT_DOUBLE ) {
dumpSamples<double>(rec, sampleCount);
}
}
template<typename T> void fillPacket(Packet *packet, uint16_t version) {
PacketDataHeader header;
header.packetType = packet->record->packetType();
header.setUOM(packet->uom.c_str());
T packetHeader;
uint32_t size = packetHeader.dataSize();
size_t dataSize = packet->record->dataSize();
size += header.dataSize() + dataSize +
packet->networkCode.size() +
packet->stationCode.size() +
packet->locationCode.size() +
packet->channelCode.size();
Plugin::BufferPtr buf = Plugin::BufferPtr(new Plugin::Buffer());
buf->resize(size);
arraybuf abuf(buf->data(), buf->size());
packetHeader.size = dataSize;
packetHeader.SIDSize[NetworkCode] = packet->networkCode.size();
packetHeader.SIDSize[StationCode] = packet->stationCode.size();
packetHeader.SIDSize[LocationCode] = packet->locationCode.size();
packetHeader.SIDSize[ChannelCode] = packet->channelCode.size();
header.version = version;
header.put(abuf);
packetHeader.put(abuf);
abuf.sputn(packet->networkCode.c_str(), packetHeader.SIDSize[NetworkCode]);
abuf.sputn(packet->stationCode.c_str(), packetHeader.SIDSize[StationCode]);
abuf.sputn(packet->locationCode.c_str(), packetHeader.SIDSize[LocationCode]);
abuf.sputn(packet->channelCode.c_str(), packetHeader.SIDSize[ChannelCode]);
packet->record->put(abuf);
packet->buffer = buf;
}
void insertPacket(Plugin::BackfillingBuffer &buf, PacketPtr packet) {
// Brute-force algorithm, should be replaced by something smarter such as
// binary search. A list might be the wrong container for that.
Plugin::BackfillingBuffer::iterator it;
size_t pos = 0;
for ( it = buf.begin(); it != buf.end(); ++it, ++pos ) {
if ( (*it)->record->endTime() > packet->record->endTime() ) {
buf.insert(it, packet);
CAPS_DEBUG("Backfilling buffer: packet inserted at pos: %zu, new "
"size: %zu, time window: %s~%s",
pos, buf.size(),
buf.front()->record->startTime().iso().c_str(),
buf.back()->record->endTime().iso().c_str());
//dump(buf);
return;
}
}
buf.push_back(packet);
CAPS_DEBUG("Backfilling buffer: packet appended, new size: %zu, time window: %s~%s",
buf.size(),
buf.front()->record->startTime().iso().c_str(),
buf.back()->record->endTime().iso().c_str());
//dump(buf);
}
#endif
Time getLastSampleTime(DataRecord *rec) {
if ( !rec ) {
return Time();
}
const DataRecord::Header *header = rec->header();
if ( header->samplingFrequencyNumerator != 0 &&
header->samplingFrequencyDenominator != 0 ) {
TimeSpan samplingInterval = header->samplingFrequencyDenominator / header->samplingFrequencyNumerator;
return rec->endTime() - samplingInterval;
}
return rec->endTime();
}
}
Plugin::Plugin(const string &name, const string &options,
const string &description)
: _name(name)
, _options(options)
, _description(description)
, _bufferSize(1 << 17)
, _bytesBuffered(0)
, _isExitRequested(false)
#if !defined(CAPS_FEATURES_BACKFILLING) || CAPS_FEATURES_BACKFILLING
, _backfillingBufferSize(0)
#endif
{
_url.host = "localhost";
_url.port = 18003;
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
_lastWrite = Time::GMT();
_journalDirty = false;
_flushInterval = 10;
#endif
_closed = false;
_wasConnected = false;
_ackTimeout = 5;
_lastAckTimeout = 5;
_sendTimeout = 60;
_readTimeout = _sendTimeout;
_connectionTimeout = 30;
_encoderFactory = NULL;
_maxFutureEndTime = TimeSpan(120, 0);
#if !defined(CAPS_FEATURES_SSL) || CAPS_FEATURES_SSL
_useSSL = false;
#endif
_dumpPackets = false;
}
Plugin::~Plugin() {
if ( !_closed ) close();
setEncoderFactory(NULL);
}
void Plugin::close() {
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
_lastWrite = Time();
#endif
_closed = true;
CAPS_INFO("[%p] Stop sending data requested", static_cast<void*>(this));
CAPS_INFO("[%p] Flushing backfilling buffer", static_cast<void*>(this));
for ( StreamStates::iterator it = _states.begin();
it != _states.end(); ++it ) {
StreamState &state = it->second;
while ( !state.backfillingBuffer.empty() ) {
PacketPtr &ref_pkt = state.backfillingBuffer.front();
state.lastCommitEndTime = ref_pkt->record->endTime();
encodePacket(ref_pkt);
state.backfillingBuffer.pop_front();
}
}
CAPS_INFO("[%p] Flushing %zu encoders",
static_cast<void*>(this), _encoderItems.size());
flushEncoders();
sendBye();
CAPS_INFO("[%p] Closing connection to CAPS at %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
while ( !_packetBuffer.empty() && readResponse(_lastAckTimeout) );
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
updateJournal();
#endif
disconnect();
CAPS_INFO("[%p] Closed connection to CAPS at %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
_packetBuffer.clear();
_wasConnected = false;
}
void Plugin::quit() {
CAPS_INFO("[%p] Shutdown was requested from outside", static_cast<void*>(this));
_isExitRequested = true;
}
bool Plugin::connect() {
#if !defined(CAPS_FEATURES_SSL) || CAPS_FEATURES_SSL
_socket = SocketPtr(_useSSL ? new CAPS::SSLSocket() : new CAPS::Socket());
#else
_socket = SocketPtr(new CAPS::Socket());
#endif
CAPS_INFO("[%p] Attempting to connect to CAPS at %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
auto status = _socket->connect(_url.host, _url.port, _connectionTimeout);
if ( status != Socket::Success ) {
CAPS_ERROR("[%p] Connection failed to CAPS at %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
return false;
}
// Do handshake
int apiVersion = 0;
if ( !getAPIVersion(apiVersion) ) {
return false;
}
CAPS_INFO("[%p] Found CAPS API version %d", static_cast<void*>(this), apiVersion);
if ( apiVersion >= 5 ) {
if ( !_hostInfo.agent.empty() ) {
_socket->send("AGENT ");
_socket->send(_hostInfo.agent.data());
_socket->send("\n");
}
}
if ( apiVersion >= 7 ) {
if ( !_hostInfo.agentVersion.empty() ) {
_socket->send("VERSION ");
_socket->send(_hostInfo.agentVersion.data());
_socket->send("\n");
}
if ( !_hostInfo.os.empty() ) {
_socket->send("OS ");
_socket->send(_hostInfo.os.data());
_socket->send("\n");
}
if ( _hostInfo.totalDisc != -1 ) {
_socket->send("TOTALDISC ");
_socket->send(toString(_hostInfo.totalDisc).c_str());
_socket->send("\n");
}
if ( _hostInfo.totalMem != -1 ) {
_socket->send("TOTALMEM ");
_socket->send(toString(_hostInfo.totalMem).c_str());
_socket->send("\n");
}
sendRuntimeInfo();
if ( !getConnectionID() ) {
return false;
}
}
if ( !_url.user.empty() && !_url.password.empty() ) {
_socket->write("AUTH ", 5);
_socket->write(_url.user.data(), _url.user.size());
_socket->write(" ", 1);
_socket->write(_url.password.data(), _url.password.size());
_socket->write("\n", 1);
}
_socket->setNonBlocking(true);
FD_ZERO(&_readFDs);
FD_ZERO(&_writeFDs);
FD_SET(_socket->fd(), &_readFDs);
FD_SET(_socket->fd(), &_writeFDs);
CAPS_INFO("[%p] Connected to CAPS at %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
_responseBuf[0] = '\0';
_responseBufIdx = 0;
bool result = true;
// Disable packet flushing for the first connection establishment to
// avoid duplicate records
if ( _wasConnected && !flush() ) {
disconnect();
result = false;
}
_wasConnected = true;
if ( result && _connectedCallback ) {
_connectedCallback();
}
return result;
}
void Plugin::disconnect() {
if ( _socket && _socket->isValid() ) {
CAPS_INFO("[%p] Disconnect from %s:%d",
static_cast<void*>(this), _url.host.c_str(), _url.port);
_socket->shutdown();
_socket->close();
_responseBuf[0] = '\0';
_responseBufIdx = 0;
if ( _disconnectedCallback ) {
_disconnectedCallback();
}
}
}
bool Plugin::isConnected() const {
return _socket && _socket->isValid();
}
bool Plugin::readResponse(unsigned int timeout) {
if ( !_socket || !_socket->isValid() ) return false;
_socket->setNonBlocking(true);
bool gotResponse = false;
const int bufN = 512;
char buf[bufN];
int res;
if ( timeout > 0 ) {
struct timeval tv;
tv.tv_sec = timeout;
tv.tv_usec = 0;
FD_ZERO(&_readFDs);
FD_SET(_socket->fd(), &_readFDs);
res = select(_socket->fd() + 1, &_readFDs, NULL, NULL, &tv);
if ( res <= 0 )
return gotResponse;
}
while ( true ) {
res = _socket->read(buf, bufN);
if ( res < 0 ) {
if ( errno != EAGAIN && errno != EWOULDBLOCK ) {
CAPS_ERROR("[%p] Reading failed: %s: disconnect",
static_cast<void*>(this), strerror(errno));
disconnect();
}
break;
}
else if ( res == 0 ) {
CAPS_INFO("[%p] Peer closed connection", static_cast<void*>(this));
disconnect();
break;
}
char *in = buf;
// Parse the input buffer into the line buffer
for ( int i = 0; i < res; ++i, ++in ) {
if ( *in == '\n' ) {
_responseBuf[_responseBufIdx] = '\0';
// Handle line
if ( (strncasecmp(_responseBuf, "OK ", 3) == 0) && (_responseBufIdx > 3) ) {
// Got OK response
// Read confirmed packets from response
int count = atoi(_responseBuf+3);
CAPS_DEBUG("[%p] Acknowledged %d packets, %zu in queue",
static_cast<void*>(this), count, _packetBuffer.size());
// Update packet buffer
for ( int i = 0; i < count; ++i ) {
if ( _packetBuffer.empty() ) {
CAPS_ERROR("[%p] Synchronization error: more packages acknowledged than in queue",
static_cast<void*>(this));
break;
}
PacketPtr packet = _packetBuffer.front();
if ( packet->record->endTime() > _states[packet->streamID].lastEndTime ) {
_states[packet->streamID].lastEndTime = packet->record->endTime();
}
_packetBuffer.pop_front();
_bytesBuffered -= packet->record->dataSize();
_packetBufferDirty = true;
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
_journalDirty = true;
#endif
gotResponse = true;
CAPS_DEBUG("[%p] Packet acknowledged by CAPS, stream: %s' time window: %s~%s",
static_cast<void*>(this),
packet->streamID.c_str(), packet->record->startTime().iso().c_str(),
packet->record->endTime().iso().c_str());
if ( _packetAckFunc ) {
_packetAckFunc(packet->streamID, packet->record->startTime(), packet->record->endTime(), packet->context);
}
}
CAPS_DEBUG("[%p] Packet buffer state: %zu packets, %zu bytes",
static_cast<void*>(this),
_packetBuffer.size(), _bytesBuffered);
}
else if ( (strncasecmp(_responseBuf, "ERROR ", 6) == 0) && (_responseBufIdx > 6) ) {
CAPS_ERROR("[%p] %s", static_cast<void*>(this), _responseBuf + 6);
return false;
}
_responseBuf[0] = '\0';
_responseBufIdx = 0;
}
else {
_responseBuf[_responseBufIdx] = *in;
++_responseBufIdx;
}
}
if ( _packetBuffer.empty() ) break;
}
return gotResponse;
}
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
void Plugin::updateJournal() {
Time currentTime = Time::GMT();
if ( !_journalFile.empty() && _journalDirty &&
_lastWrite + TimeSpan(_flushInterval) <= currentTime ) {
writeJournal();
_journalDirty = false;
_lastWrite = currentTime;
}
}
bool Plugin::readJournal() {
if ( _journalFile.empty() ) return false;
ifstream ifs;
ifs.open(_journalFile.c_str());
return readJournal(ifs);
}
bool Plugin::readJournal(istream &is) {
_states.clear();
string line;
Time time;
int lineNumber = 0;
Time now = Time::GMT();
Time maxAllowedEndTime = now + TimeSpan(86400);
while ( getline(is, line) ) {
++lineNumber;
trim(line);
if ( line.empty() ) {
continue;
}
if ( line[0] == '#' ) {
continue;
}
std::vector<std::string> toks;
boost::split(toks, line, boost::is_any_of(" "));
if ( toks.empty() ) {
CAPS_ERROR("[%p] journal:%d: Invalid line: %s",
static_cast<void*>(this), lineNumber, line.c_str());
return false;
}
std::string streamID = toks[0];
string strTime;
if ( toks.size() > 1 ) {
strTime = toks[1];
}
if ( !strTime.empty() ) {
if ( !time.fromString(strTime.c_str(), "%FT%T.%fZ") ) {
CAPS_ERROR("[%p] journal:%d: Invalid time: %s",
static_cast<void*>(this), lineNumber, strTime.c_str());
return false;
}
}
else {
time = Time();
}
if ( time > maxAllowedEndTime ) {
CAPS_WARNING("[%p] journal:%d:%s: Timestamp %s is more than one day "
"ahead of current time, respecting it nevertheless.",
static_cast<void*>(this),
lineNumber, streamID.c_str(), time.iso().c_str());
}
_states[streamID].lastEndTime = time;
}
return true;
}
Gempa::CAPS::Time Plugin::lastEndTime(const std::string &id) {
Gempa::CAPS::Plugin::StreamStates::const_iterator it = _states.find(id);
if ( it == _states.end() ) {
return Gempa::CAPS::Time();
}
return it->second.lastEndTime;
}
#endif
bool Plugin::flush() {
CAPS_INFO("[%p] Flushing %zu queued packets", static_cast<void*>(this),
_packetBuffer.size());
PacketBuffer::iterator it = _packetBuffer.begin();
while ( it != _packetBuffer.end() && !_isExitRequested ) {
PacketPtr packet = *it;
_packetBufferDirty = false;
if ( !sendPacket(packet.get() ) && !_isExitRequested ) {
if ( _packetBufferDirty ) {
CAPS_ERROR("[%p] Uh oh, buffer dirty but sending failed!",
static_cast<void*>(this));
}
return false;
}
if ( readResponse() ) {
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
updateJournal();
#endif
}
if ( _packetBufferDirty ) {
it = std::find(_packetBuffer.begin(), _packetBuffer.end(), packet);
if ( it != _packetBuffer.end() )
++it;
else {
CAPS_DEBUG("[%p] Last packet removed, reset flush queue iterator",
static_cast<void*>(this));
it = _packetBuffer.begin();
}
}
else
++it;
}
return true;
}
void Plugin::flushEncoders() {
EncoderItems::iterator it = _encoderItems.begin();
while ( it != _encoderItems.end() ) {
Encoder *encoder = it->second.encoder.get();
if ( encoder ) {
encoder->flush();
PacketPtr encodedPacket;
while ( (encodedPacket = encoder->pop()) ) {
commitPacket(encodedPacket);
}
}
it++;
}
}
Plugin::Status Plugin::push(const string &net, const string &sta,
const string &loc, const string &cha,
const Time &stime,
uint16_t numerator, uint16_t denominator,
const string &uom,
void *data, size_t count, DataType dt,
int timingQuality, void *context) {
return pushRaw(net, sta, loc, cha, stime, numerator, denominator, uom,
data, count, dt, timingQuality, context);
}
Plugin::Status Plugin::pushRaw(const string &net, const string &sta,
const string &loc, const string &cha,
const Time &stime,
uint16_t numerator, uint16_t denominator,
const string &uom,
void *data, size_t count, DataType dt,
int timingQuality, void *context) {
uint8_t dtSize = dataTypeSize(dt);
if ( !dtSize ) {
return Plugin::PacketLoss;
}
RawDataRecord *rec = new RawDataRecord();
rec->setStartTime(stime);
rec->setSamplingFrequency(numerator, denominator);
rec->setDataType(dt);
rec->setBuffer(static_cast<char*>(data), dtSize * count);
return push(net, sta, loc, cha, DataRecordPtr(rec), uom, timingQuality, context);
}
#if !defined(CAPS_FEATURES_ANY) || CAPS_FEATURES_ANY
Plugin::Status Plugin::push(const string &net, const std::string &sta,
const std::string &loc, const std::string &cha,
const Time &stime, uint16_t numerator,
uint16_t denominator, const std::string &format,
char *data, size_t count, void *context) {
AnyDataRecord *rec = new AnyDataRecord;
rec->setStartTime(stime);
rec->setEndTime(stime);
rec->setSamplingFrequency(numerator, denominator);
rec->setType(format.c_str());
rec->setData(data, count);
return push(net, sta, loc, cha, DataRecordPtr(rec), "px", -1, context);
}
Plugin::Status Plugin::pushAny(const string &net, const std::string &sta,
const std::string &loc, const std::string &cha,
const Time &stime, const Time &etime,
uint16_t numerator, uint16_t denominator,
const std::string &format, const std::string &uom,
char *data, size_t count, void *context) {
AnyDataRecord *rec = new AnyDataRecord;
rec->setStartTime(stime);
rec->setEndTime(etime);
rec->setSamplingFrequency(numerator, denominator);
rec->setType(format.c_str());
rec->setData(data, count);
return push(net, sta, loc, cha, DataRecordPtr(rec), uom, -1, context);
}
Plugin::Status Plugin::push(const string &net, const std::string &sta,
const std::string &loc, const std::string &cha,
const Time &stime, uint16_t numerator,
uint16_t denominator, const std::string &format,
const std::string &str, void *context) {
return push(net, sta, loc, cha, stime, numerator, denominator, format,
const_cast<char*>(str.data()), str.size(), context);
}
Plugin::Status Plugin::pushAny(const string &net, const std::string &sta,
const std::string &loc, const std::string &cha,
const Time &stime, const Time &etime,
uint16_t numerator, uint16_t denominator,
const std::string &format, const std::string &uom,
const std::string &str, void *context) {
return pushAny(net, sta, loc, cha, stime, etime,
numerator, denominator, format, uom,
const_cast<char*>(str.data()), str.size(), context);
}
#endif
void Plugin::tryFlushBackfillingBuffer(StreamState &state) {
if ( state.backfillingBuffer.empty() )
return;
size_t flushed = 0;
while ( !state.backfillingBuffer.empty() ) {
PacketPtr &ref_pkt = state.backfillingBuffer.front();
TimeSpan gap = ref_pkt->record->startTime() - state.lastCommitEndTime;
int64_t dt_us = static_cast<int64_t>(gap.seconds()) * 1000000 + gap.microseconds();
// A gap larger than one sample?
if ( dt_us >= ref_pkt->dt_us ) break;
state.lastCommitEndTime = ref_pkt->record->endTime();
encodePacket(ref_pkt);
state.backfillingBuffer.pop_front();
++flushed;
//dump(state.backfillingBuffer);
}
if ( flushed > 0 ) {
if ( state.backfillingBuffer.size() > 0 ) {
CAPS_DEBUG("[%p] backfilling buffer: %zu pakets flushed, new size: %zu, time window: %s~%s",
static_cast<void*>(this),
flushed, state.backfillingBuffer.size(),
state.backfillingBuffer.front()->record->startTime().iso().c_str(),
state.backfillingBuffer.back()->record->endTime().iso().c_str());
}
else {
CAPS_DEBUG("[%p] backfilling buffer: %zu pakets flushed, new size: 0",
static_cast<void*>(this),
flushed);
}
}
}
void Plugin::trimBackfillingBuffer(StreamState &state) {
if ( state.backfillingBuffer.empty() )
return;
size_t trimmed = 0;
while ( !state.backfillingBuffer.empty() ) {
TimeSpan diff = state.backfillingBuffer.back()->record->endTime() -
state.backfillingBuffer.front()->record->startTime();
if ( diff.seconds() <= _backfillingBufferSize )
break;
PacketPtr &ref_pkt = state.backfillingBuffer.front();
state.lastCommitEndTime = ref_pkt->record->endTime();
encodePacket(ref_pkt);
state.backfillingBuffer.pop_front();
}
if ( trimmed > 0 ) {
if ( state.backfillingBuffer.size() > 0 ) {
CAPS_DEBUG("[%p] backfilling buffer: %zu pakets trimmed, new size: %zu, time window: %s~%s",
static_cast<void*>(this),
trimmed, state.backfillingBuffer.size(),
state.backfillingBuffer.front()->record->startTime().iso().c_str(),
state.backfillingBuffer.back()->record->endTime().iso().c_str());
}
else {
CAPS_DEBUG("[%p] backfilling buffer: %zu pakets trimmed, new size: 0",
static_cast<void*>(this), trimmed);
}
}
}
Plugin::Status Plugin::push(const string &net, const string &sta,
const string &loc, const string &cha,
DataRecordPtr rec, const string &uom,
int timingQuality, void *context) {
return pushRecord(net, sta, loc, cha, rec, uom, timingQuality, context);
}
Plugin::Status Plugin::pushRecord(const string &net, const string &sta,
const string &loc, const string &cha,
DataRecordPtr rec, const string &uom,
int timingQuality, void *context) {
static bool showVersion = true;
if ( showVersion ) {
CAPS_NOTICE("[%p] LIB CAPS version %s", static_cast<void*>(this), version());
showVersion = false;
}
if ( !rec ) {
return PacketNotValid;
}
if ( rec->dataSize() > _bufferSize ) {
return PacketSize;
}
if ( !rec->endTime().valid() ) {
return PacketNotValid;
}
if ( !validateStreamID(net, sta, loc, cha) ) {
return PacketNotValid;
}
if ( _maxFutureEndTime.seconds() >= 0 ) {
Time endTime = getLastSampleTime(rec.get());
Time maxAllowedEndTime = Time::GMT() + TimeSpan(_maxFutureEndTime);
if ( endTime > maxAllowedEndTime ) {
CAPS_WARNING("[%p] %s.%s.%s.%s: Future time stamp detected: Packet end time %s exceeds "
"max allowed end time %s. Discard packet.",
static_cast<void*>(this),
net.c_str(), sta.c_str(), loc.c_str(), cha.c_str(),
rec->endTime().iso().c_str(),
maxAllowedEndTime.iso().c_str());
return MaxFutureEndTimeExceeded;
}
}
const DataRecord::Header *header = rec->header();
PacketPtr packet(new Packet(rec, net, sta, loc, cha, context));
packet->uom = uom;
packet->dataType = header->dataType;
packet->timingQuality = timingQuality;
#if !defined(CAPS_FEATURES_BACKFILLING) || CAPS_FEATURES_BACKFILLING
if ( header->samplingFrequencyNumerator == 0 ) {
CAPS_DEBUG("[%p] detected division by zero, invalid sampling frequency numerator",
static_cast<void*>(this));
return PacketNotValid;
}
packet->dt_us = ((int64_t)header->samplingFrequencyDenominator * 1000000) /
header->samplingFrequencyNumerator;
#endif
StreamState &state = _states[packet->streamID];
#if !defined(CAPS_FEATURES_BACKFILLING) || CAPS_FEATURES_BACKFILLING
// If buffering is enabled
if ( _backfillingBufferSize > 0 ) {
if ( state.lastCommitEndTime.valid() ) {
int64_t dt_us = ((int64_t)rec->header()->samplingFrequencyDenominator * 1000000) /
rec->header()->samplingFrequencyNumerator;
TimeSpan gap = packet->record->startTime() - state.lastCommitEndTime;
// A gap larger than one sample?
if ( ((int64_t)gap.seconds()*1000000+gap.microseconds()) >= dt_us ) {
CAPS_DEBUG("[%p] detected gap on stream: %s",
static_cast<void*>(this), packet->streamID.c_str());
insertPacket(state.backfillingBuffer, packet);
trimBackfillingBuffer(state);
tryFlushBackfillingBuffer(state);
return Success;
}
}
if ( !encodePacket(packet) )
return PacketLoss;
if ( rec->endTime() > state.lastCommitEndTime) {
state.lastCommitEndTime = rec->endTime();
}
tryFlushBackfillingBuffer(state);
}
else {
#endif
if ( !encodePacket(packet) ) {
return PacketLoss;
}
#if !defined(CAPS_FEATURES_BACKFILLING) || CAPS_FEATURES_BACKFILLING
}
#endif
return Success;
}
void Plugin::serializePacket(Packet *packet) {
if ( packet->buffer ) return;
size_t dataSize = packet->record->dataSize();
bool isV1 = dataSize < (1 << 16);
if ( isV1 ) {
fillPacket<PacketHeaderV1>(packet, 1);
}
else {
fillPacket<PacketHeaderV2>(packet, 2);
}
}
bool Plugin::commitPacket(PacketPtr packet) {
if ( _dumpPackets ) {
serializePacket(packet.get());
dumpPacket(packet.get());
return true;
}
// Initial connect
if ( !_wasConnected && !_socket ) {
while ( !connect() && !_isExitRequested ) {
disconnect();
wait();
}
if ( !isConnected() ) {
return false;
}
}
if ( _bytesBuffered >= _bufferSize ) {
CAPS_DEBUG("[%p] Packet buffer is full (%zu/%zu bytes), "
"waiting for server ack messages",
static_cast<void*>(this),
_bytesBuffered, _bufferSize);
while ( _bytesBuffered >= _bufferSize ) {
if ( !readResponse(_ackTimeout) ) {
CAPS_WARNING("[%p] Packet buffer was full (%zu/%zu bytes), "
"did not receive ack within %d seconds: %s",
static_cast<void*>(this),
_bytesBuffered, _bufferSize,
_ackTimeout, _isExitRequested ? "abort" : "reconnect");
disconnect();
while ( !_isExitRequested && !_socket->isValid() ) {
wait();
disconnect();
connect();
}
if ( _isExitRequested )
break;
}
}
CAPS_DEBUG("[%p] %zu/%zu bytes buffered after force wait",
static_cast<void*>(this),
_bytesBuffered, _bufferSize);
if ( _bytesBuffered >= _bufferSize ) {
return false;
}
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
CAPS_DEBUG("[%p] Force journal update", static_cast<void*>(this));
#endif
}
// Serialize data record
serializePacket(packet.get());
CAPS_DEBUG("[%p] + buffer state: %zu packets, %zu bytes",
static_cast<void*>(this),
_packetBuffer.size(), _bytesBuffered);
while ( !sendPacket(packet.get() ) ) {
CAPS_ERROR("[%p] Sending failed: %s", static_cast<void*>(this), _isExitRequested ? "abort" : "reconnect");
readResponse();
disconnect();
while ( !_isExitRequested && !isConnected() ) {
wait();
disconnect();
connect();
}
if ( !isConnected() ) {
return false;
}
}
_packetBuffer.push_back(packet);
_bytesBuffered += packet->record->dataSize();
_stats.maxBytesBuffered = max(_bytesBuffered, _stats.maxBytesBuffered);
if ( readResponse() ) {
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
updateJournal();
#endif
}
return true;
}
Encoder* Plugin::getEncoder(PacketPtr packet) {
EncoderItems::iterator it = _encoderItems.find(packet->streamID);
if ( it == _encoderItems.end() ) {
EncoderItem item;
it = _encoderItems.insert(make_pair(packet->streamID, item)).first;
}
DataRecord *record = packet->record.get();
const DataRecord::Header *header = record->header();
EncoderItem *item = &it->second;
if ( item->encoder ) {
// Copy context
item->encoder->setContext(packet->context);
/* Before we can feed data into the encoder we
have to check if the sampling frequency
or data type have been changed. Also gaps
must be handled.
*/
const SPClock &clk = item->encoder->clk();
bool needsFlush = false;
if ( clk.freqn != header->samplingFrequencyNumerator ||
clk.freqd != header->samplingFrequencyDenominator ) {
needsFlush = true;
}
else if ( record->startTime() != clk.getTime(0) ) {
auto gap = abs(static_cast<double>(record->startTime() - clk.getTime(0)));
if ( (header->samplingFrequencyNumerator == 0)
|| (gap > header->samplingFrequencyDenominator * 0.5 / header->samplingFrequencyNumerator) ) {
needsFlush = true;
}
}
else if ( item->dataType != header->dataType ) {
needsFlush = true;
}
else if ( item->encoder->timingQuality() != packet->timingQuality ) {
needsFlush = true;
}
if ( needsFlush ) {
item->encoder->flush();
PacketPtr encodedPacket;
while ( (encodedPacket = item->encoder->pop()) ) {
// TODO check return value?
commitPacket(encodedPacket);
}
item->encoder = EncoderPtr();
}
}
if ( !item->encoder ) {
Encoder *encoder =
_encoderFactory->create(packet->networkCode, packet->stationCode,
packet->locationCode, packet->channelCode,
header->dataType,
header->samplingFrequencyNumerator,
header->samplingFrequencyDenominator);
if ( encoder ) {
encoder->setContext(packet->context);
encoder->setStartTime(record->startTime());
encoder->setTimingQuality(packet->timingQuality);
item->encoder = EncoderPtr(encoder);
item->dataType = header->dataType;
}
}
return item->encoder.get();
}
bool Plugin::encodePacket(PacketPtr packet) {
if ( !_encoderFactory ||
!_encoderFactory->supportsRecord(packet->record.get()) ) {
return commitPacket(packet);
}
Buffer *buffer = packet->record->data();
if ( !buffer ) {
return commitPacket(packet);
}
Encoder *encoder = getEncoder(packet);
if ( !encoder ) {
return commitPacket(packet);
}
uint8_t dtSize = dataTypeSize(packet->dataType);
for ( size_t i = 0; i < buffer->size(); i = i + dtSize ) {
char *data = buffer->data() + i;
encoder->push(data);
}
PacketPtr encodedPacket;
while ( (encodedPacket = encoder->pop()) ) {
// TODO check return value?
commitPacket(encodedPacket);
}
return true;
}
void Plugin::setEncoderFactory(EncoderFactory *factory) {
if ( _encoderFactory ) {
delete _encoderFactory;
_encoderFactory = NULL;
}
_encoderFactory = factory;
}
void Plugin::sendBye() {
if ( _socket == NULL ) return;
PacketDataHeader header;
memset(reinterpret_cast<char*>(&header), 0, header.dataSize());
_socket->setNonBlocking(false);
send((char*)&header, header.dataSize(), _sendTimeout);
}
bool Plugin::sendPacket(Packet *packet) {
if ( !_packetBuffer.empty() ) {
readResponse();
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
updateJournal();
#endif
}
// Send packet data header and packet header
_socket->setNonBlocking(false);
// Instead of directly write data to the socket we call
// a wrapper function that checks if we can write data or not.
// If we do not do the check the system send operation
// hangs when the TCP buffer is full and the connection is
// is in an undefined state.
// The check and the system send operation run in a loop until
// the data has been sent successfully or the user requests to
// to stop => The function call might block.
if ( !send(packet->buffer->data(), packet->buffer->size(), _sendTimeout) ) {
return false;
}
CAPS_DEBUG("[%p] Packet sent to CAPS, stream: %s, time window: %s~%s",
static_cast<void*>(this),
packet->streamID.c_str(), packet->record->startTime().iso().c_str(),
packet->record->endTime().iso().c_str());
return true;
}
void Plugin::wait() {
// Wait 5 seconds and keep response latency low
for ( int i = 0; (i < 10) && !_isExitRequested; ++i )
usleep(500000);
}
#if !defined(CAPS_FEATURES_JOURNAL) || CAPS_FEATURES_JOURNAL
bool Plugin::writeJournal() {
//if ( _states.empty() ) return true;
if ( !_journalFile.empty() ) {
fs::path path(_journalFile);
string filename = ".journal.tmp";
if ( path.has_parent_path() ) {
createPath(path.parent_path().string());
filename = (path.parent_path() / filename).string();
}
ofstream ofs(filename.c_str());
if ( !ofs.is_open() ) return false;
if ( writeJournal(ofs) ) {
ofs.close();
if ( path.has_parent_path() ) {
createPath(path.parent_path().string());
}
if ( rename(filename.c_str(), path.string().c_str()) != 0 )
CAPS_ERROR("[%p] Failed to create journal %s, %s(%d)",
static_cast<void*>(this), _journalFile.c_str(),
strerror(errno), errno);
}
else {
ofs.close();
try {
fs::remove(filename);
}
catch ( ... ) {}
}
}
else
return writeJournal(cout);
return false;
}
bool Plugin::writeJournal(ostream &os) {
for ( StreamStates::iterator it = _states.begin();
it != _states.end(); ++it ) {
os << it->first << " " << it->second.lastEndTime.iso() << endl;
if ( !os.good() ) return false;
}
return true;
}
#endif
void Plugin::enableLogging() {
Gempa::CAPS::SetLogHandler(Gempa::CAPS::LL_ERROR, LogError);
Gempa::CAPS::SetLogHandler(Gempa::CAPS::LL_WARNING, LogWarning);
Gempa::CAPS::SetLogHandler(Gempa::CAPS::LL_NOTICE, LogNotice);
Gempa::CAPS::SetLogHandler(Gempa::CAPS::LL_INFO, LogInfo);
Gempa::CAPS::SetLogHandler(Gempa::CAPS::LL_DEBUG, LogDebug);
}
bool Plugin::send(char *data, int len, int timeout) {
if ( !_socket->isValid() ) {
return false;
}
struct timeval tv;
tv.tv_sec = timeout;
tv.tv_usec = 0;
FD_ZERO(&_writeFDs);
FD_SET(_socket->fd(), &_writeFDs);
int res = select(_socket->fd() + 1, NULL, &_writeFDs, NULL, &tv);
if ( res > 0 ) {
res = _socket->write(data, len);
if ( res == len ) {
return true;
}
else if ( res == -1 ) {
CAPS_ERROR("[%p] Failed to send data: %s. "
"Forcing reconnect and trying to send data again.",
static_cast<void*>(this), strerror(errno));
}
else {
CAPS_ERROR("[%p] Incomplete send operation: "
"Only %d/%d bytes have been sent. "
"Forcing reconnect and trying to send data again.",
static_cast<void*>(this), len, res);
}
}
else if ( !res ) {
CAPS_ERROR("[%p] Detected hanging TCP connection. "
"Forcing reconnect and trying to send data again.",
static_cast<void*>(this));
}
else {
CAPS_ERROR("[%p] Send error: %s", static_cast<void*>(this), strerror(errno));
}
return false;
}
bool Plugin::setAddress(const string &addr, uint16_t defaultPort) {
if ( !_url.parse(addr, defaultPort) ) {
CAPS_ERROR("[%p] Failed to set address: %s",
static_cast<void*>(this), _url.errorString.c_str());
return false;
}
#if !defined(CAPS_FEATURES_SSL) || CAPS_FEATURES_SSL
_useSSL = boost::iequals(_url.protocol, "capss");
#endif
return true;
}
void Plugin::dumpPacket(Packet *packet) {
if ( packet->record->packetType() == RawDataPacket ) {
dumpSList(packet);
}
else if ( packet->record->packetType() == MSEEDPacket ) {
MSEEDDataRecord *rec = static_cast<MSEEDDataRecord*>(packet->record.get());
cout.write(rec->data()->data(),
static_cast<streamsize>(rec->data()->size()));
}
else {
cout << "Could not dump packet: Unsupported packet type '"
<< packet->record->packetType() << "'";
}
}
void Plugin::dumpPackets(bool enable) {
_dumpPackets = enable;
}
bool Plugin::getAPIVersion(int &version) {
version = 0;
int bytesSent = _socket->send(HELLO_REQUEST);
if ( bytesSent != strlen(HELLO_REQUEST) ) {
CAPS_ERROR("[%p] Could not get CAPS API version: %s",
static_cast<void*>(this), strerror(errno));
return false;
}
int32_t msgLength = 0;
socketbuf<Socket, 512> buf(_socket.get());
streamsize bytesRead = buf.sgetn(reinterpret_cast<char*>(&msgLength),
sizeof(int32_t));
if ( bytesRead != sizeof(int32_t) ) {
CAPS_ERROR("[%p] Could not get CAPS API version: Expected message length in "
"server response: %s", static_cast<void*>(this), strerror(errno));
return false;
}
msgLength = Endianess::Converter::FromLittleEndian(msgLength);
// Check if the hello response comes from a CAPS server with
// API version < 5
constexpr int32_t OKTag = 0x30204b4f;
if ( msgLength == OKTag ) {
return true;
}
buf.set_read_limit(msgLength);
bool ret = true;
iostream is(&buf);
string line;
while ( getline(is, line) ) {
size_t pos = line.find("API=");
if ( pos == string::npos ) {
continue;
}
string apiStr = line.substr(4);
if ( !str2int(version, apiStr.c_str()) ) {
CAPS_ERROR("[%p] Invalid CAPS API version: Expected number", static_cast<void*>(this));
ret = false;
}
}
return ret;
}
void Plugin::setConnectionID(const std::string &id) {
_connectionID = id;
}
bool Plugin::setConnectionIDFile(const std::string &filename) {
_connectionIDFile = filename;
if ( fs::exists(_connectionIDFile) ) {
std::ifstream ifs(_connectionIDFile.c_str());
if ( !ifs.is_open() ) {
CAPS_ERROR("[%p] Could not open connection ID file %s",
static_cast<void*>(this), _connectionIDFile.c_str());
return false;
}
ifs >> _connectionID;
if ( ifs.fail() ) {
CAPS_ERROR("[%p] Could not read connection ID from file %s",
static_cast<void*>(this), _connectionIDFile.c_str());
return false;
}
}
return true;
}
const std::string &Plugin::connectionID() const {
return _connectionID;
}
bool Plugin::getConnectionID() {
_socket->send("ID ");
_socket->send(_connectionID.data());
_socket->send("\n");
try {
uint32_t size = 0;
auto res = _socket->read(reinterpret_cast<char*>(&size), sizeof(size));
if ( res != sizeof(size) ) {
throw std::runtime_error("Expected connection ID response");
}
size = Endianess::Converter::FromLittleEndian(size);
auto bytesRead = _socket->read(_responseBuf, size);
if ( bytesRead != static_cast<std::streamsize>(size) ) {
throw std::runtime_error("Server returned not connection ID");
}
_responseBuf[size] = '\0';
if ( strncmp(_responseBuf, "ID ", 3 ) != 0 ) {
throw std::runtime_error("Invalid connection ID response");
return false;
}
_connectionID.assign(_responseBuf + 3, size - 3);
CAPS_INFO("[%p] CAPS connection ID %s", static_cast<void*>(this),
_connectionID.data());
if ( !_connectionIDFile.empty() ) {
fs::path path(_connectionIDFile);
std::string filename = "connection";
if ( path.has_parent_path() ) {
auto parentPath = path.parent_path().string();
if ( !createPath(parentPath) ) {
runtime_error("Could not create directory " + parentPath);
}
}
std::ofstream ofs(_connectionIDFile);
if ( !ofs.is_open() ) {
throw std::runtime_error("Could not create file " + _connectionIDFile);
}
ofs << _connectionID;
if ( !ofs.good() ) {
throw std::runtime_error("Failed to write connection id to file "
+ _connectionIDFile);
}
}
}
catch ( const std::runtime_error &err ) {
CAPS_ERROR("[%p] Connection failed to CAPS at %s:%d: ",
static_cast<void*>(this), _url.host.data(), _url.port,
err.what());
return false;
}
return true;
}
void Plugin::setHostInfo(const HostInfo &info) {
_hostInfo = info;
}
void Plugin::setRuntimeInfo(const RuntimeInfo &info) {
_runtimeInfo = info;
sendRuntimeInfo();
}
void Plugin::setConnectedCallback(ConnectedCallback cb) {
_connectedCallback = std::move(cb);
}
void Plugin::setDisconnectedCallback(DisconnectedCallback cb) {
_disconnectedCallback = std::move(cb);
}
void Plugin::sendRuntimeInfo() {
if ( !_wasConnected && !_socket ) {
while ( !connect() && !_isExitRequested ) {
disconnect();
wait();
}
if ( !isConnected() ) {
return;
}
}
std::string buffer;
if ( _runtimeInfo.cpuUsage != -1 ) {
buffer += "CPU ";
buffer += toString(_runtimeInfo.cpuUsage);
buffer += "\n";
}
if ( _runtimeInfo.procUsedMem != -1 ) {
buffer += "MEM ";
buffer += toString(_runtimeInfo.procUsedMem);
buffer += "\n";
}
if ( _runtimeInfo.availableDisc != -1 ) {
buffer += "AVAILABLEDISC ";
buffer += toString(_runtimeInfo.availableDisc);
buffer += "\n";
}
if ( _runtimeInfo.availableMem != -1 ) {
buffer += "AVAILABLEMEM ";
buffer += toString(_runtimeInfo.availableMem);
buffer += "\n";
}
if ( _runtimeInfo.systemLoad != -1 ) {
buffer += "LOAD ";
buffer += toString(_runtimeInfo.systemLoad);
buffer += "\n";
}
// Send packet data header and packet header
_socket->setNonBlocking(false);
// Instead of directly write data to the socket we call
// a wrapper function that checks if we can write data or not.
// If we do not do the check the system send operation
// hangs when the TCP buffer is full and the connection is
// is in an undefined state.
// The check and the system send operation run in a loop until
// the data has been sent successfully or the user requests to
// to stop => The function call might block.
while ( !send(buffer.data(), buffer.size(), _sendTimeout) ) {
CAPS_ERROR("[%p] Sending rti failed: %s", static_cast<void*>(this), _isExitRequested ? "abort" : "reconnect");
readResponse();
disconnect();
while ( !_isExitRequested && !isConnected() ) {
wait();
disconnect();
connect();
}
if ( !isConnected() ) {
return;
}
}
}
}
}
Reference in New Issue View Git Blame Copy Permalink