/** * WebSocketServer to publish Events * $Id$ * $HeadURL$ * $Author$ * Created by Christian Ehrlicher, SLSE, 2017-02-06 */ #include "web_socket_server.h" #include #include #include #include #include #include #include #include #include namespace ExerciseEventManagement { WebSocketEventServer::WebSocketEventServer(const QSqlDatabase &db, QObject *parent) : QObject(parent) , m_wsServer(nullptr) , m_cleanupTimer(new QTimer(this)) , m_db(db) , m_logger(getLogger("WebSocketEventServer")) { m_cleanupTimer->setInterval(10000); // clear every 10s connect(m_cleanupTimer, &QTimer::timeout, this, &WebSocketEventServer::onCleanupLiveHistory); } WebSocketEventServer::~WebSocketEventServer() { for (auto s : qAsConst(m_liveSockets)) { s->disconnect(this); s->close(); } qDeleteAll(m_liveSockets); } bool WebSocketEventServer::listen(const QPair &portRange) { for (quint16 currentPort = portRange.first; currentPort <= portRange.second; ++currentPort) { if (listen(currentPort)) return true; } return listen(0); } bool WebSocketEventServer::listen(quint16 port) { if (m_wsServer && m_wsServer->isListening()) return true; m_cleanupTimer->start(); delete m_wsServer; m_wsServer = new QWebSocketServer(QStringLiteral("EEM-Events"), QWebSocketServer::NonSecureMode, this); connect(m_wsServer, &QWebSocketServer::newConnection, this, &WebSocketEventServer::onNewConnection); bool bRet = m_wsServer->listen(QHostAddress::Any, port); if (bRet) NOTICE_LOG(m_logger) << "Listening on port " << listenPort(); return bRet; } quint16 WebSocketEventServer::listenPort() const { return m_wsServer ? m_wsServer->serverPort() : 0; } void WebSocketEventServer::sendEvent(const QJsonObject &obj) { const QJsonDocument doc(obj); const QString msg = QString::fromUtf8(doc.toJson(QJsonDocument::Compact)); for (QWebSocket *ws : qAsConst(m_liveSockets)) ws->sendTextMessage(msg); m_liveHistory[time(nullptr)].push_back(msg); } void WebSocketEventServer::onNewConnection() { QWebSocket *ws = m_wsServer->nextPendingConnection(); if (!ws) return; DEBUG_LOG(m_logger) << "New incoming connection from " << ws->peerAddress().toString() << ":" << ws->peerPort() << ", " << ws->requestUrl().toString(); const QUrl requestUrl = ws->requestUrl(); const QString requestPath = requestUrl.path(); if (requestPath == QStringLiteral("/live")) { connect(ws, &QWebSocket::disconnected, this, &WebSocketEventServer::onSocketDisconnected); for (const auto &historyVec : qAsConst(m_liveHistory)) { for (const auto &msg : historyVec) ws->sendTextMessage(msg); } m_liveSockets += ws; } else if (requestPath.startsWith(QLatin1String("/request"))) historyDataRequested(m_db, ws, requestUrl); else { WARNING_LOG(m_logger) << "Unknown request path: '" << requestPath << "' - closing"; ws->close(); ws->deleteLater(); } } void WebSocketEventServer::onSocketDisconnected() { QWebSocket *ws = qobject_cast(sender()); if (!ws) return; DEBUG_LOG(m_logger) << "Disconnected from " << ws->peerAddress().toString() << ":" << ws->peerPort() << ", " << ws->requestUrl().toString(); ws->deleteLater(); m_liveSockets -= ws; } void WebSocketEventServer::onCleanupLiveHistory() { qint64 tNow = time(nullptr) - 120; // 2min auto it = m_liveHistory.begin(); while (it != m_liveHistory.end() && it.key() < tNow) it = m_liveHistory.erase(it); } void WebSocketEventServer::historyDataRequested(const QSqlDatabase &db, QWebSocket *ws, const QUrl &requestUrl) { const QUrlQuery query(requestUrl); const int requestId = query.queryItemValue(QStringLiteral("requestId")).toInt(); const qint64 from = query.queryItemValue(QStringLiteral("from")).toLongLong(); const qint64 to = query.queryItemValue(QStringLiteral("to")).toLongLong(); if (requestId == 0 || from <= 0 || to <= 0 || from > to) { WARNING_LOG(m_logger) << "Unknown request: '" << requestUrl.toString() << "' - closing, " << requestId << ", " << from << ", to " << to; ws->sendTextMessage(QStringLiteral("{}")); ws->close(); ws->deleteLater(); return; } QString stmt(QLatin1String(" SELECT id, event_type_id, participant_id, party_id, equipment_number, properties, sample_time, position_x, position_y, position_z" " FROM eem_event WHERE sample_time > ") + QString::number(from) + QLatin1String(" AND sample_time <= ") + QString::number(to)); // add event type filter to query if preset if (query.hasQueryItem(QStringLiteral("eventTypeId"))) { const QString ids = query.queryItemValue(QStringLiteral("eventTypeId")); // split/unsplit to make sure only integers are there const QVector idVec = ids.splitRef(QLatin1Char(',')); QStringList eventList; eventList.reserve(idVec.size()); for (const auto &id : idVec) { int i = id.toInt(); if (i == 0) continue; eventList += QString::number(i); } const QString eventStr = eventList.join(QLatin1Char(',')); stmt += QStringLiteral(" AND event_type_id IN (") + eventStr + QLatin1Char(')'); } // add Json properties filter to query if present if (query.hasQueryItem(QStringLiteral("filter"))) { const QString filterString = query.queryItemValue(QStringLiteral("filter")); DEBUG_LOG(m_logger) << "filter string: " << filterString; const QJsonObject obj(QJsonDocument::fromJson(filterString.toUtf8()).object()); if (obj.keys().size() == 1) { const QJsonValue val = obj.value(obj.keys().constFirst()); if (val.isObject()) stmt += QStringLiteral(" AND properties @> '") + filterString + QLatin1Char('\''); else if (val.isArray()) { // currently we only support int arrays const QJsonArray arr(val.toArray()); QStringList sl; sl.reserve(arr.size()); for (const auto &ele : arr) sl += QString::number(ele.toInt()); if (sl.size() > 0) stmt += QStringLiteral(" AND CAST (properties ->> '%1' AS int) IN (").arg(obj.keys().constFirst()) + sl.join(QLatin1Char(',')) + QLatin1Char(')'); } } } // add a limit for max query rows if (query.hasQueryItem(QStringLiteral("limit"))) { const QString limitString = query.queryItemValue(QStringLiteral("limit")); DEBUG_LOG(m_logger) << "limiting query to " << limitString << " events"; stmt += QStringLiteral(" LIMIT ") + limitString; } DEBUG_LOG(m_logger) << "stmt: " << stmt; NSqlQuery q(db); if (!q.exec(stmt)) { ERROR_LOG(m_logger) << "Error while selecting event-data: " << stmt << ", " << q.lastError().databaseText(); ws->sendTextMessage(QStringLiteral("{}")); ws->close(); ws->deleteLater(); return; } QJsonArray events; Database::EemEvent data; while (q.next()) { int i = 0; data.id = q.value(i++).toInt(); data.eventTypeId = static_cast(q.value(i++).toInt()); data.participantId = q.value(i++).toInt(); data.partyId = q.value(i++).toInt(); data.equipmentNumber = q.value(i++).toInt(); data.properties = q.value(i++).toString(); data.sampleTime = q.value(i++).toLongLong(); double pos_lon = q.value(i++).toDouble(); double pos_lat = q.value(i++).toDouble(); double pos_z = q.value(i++).toDouble(); data.position = LtsUtils::QPoint3D(pos_lon, pos_lat, pos_z); events.append(Database::toJson(data)); if (events.size() > 100) { INFO_LOG(m_logger) << "Sending an answer with " << events.size() << " events for query " << stmt; const QJsonDocument doc(events); ws->sendTextMessage(QString::fromUtf8(doc.toJson(QJsonDocument::Compact))); events = QJsonArray(); } } if (!events.isEmpty()) { INFO_LOG(m_logger) << "Sending an answer with " << events.size() << " events for query " << stmt; const QJsonDocument doc(events); ws->sendTextMessage(QString::fromUtf8(doc.toJson(QJsonDocument::Compact))); } // finished ws->sendTextMessage(QStringLiteral("{}")); } }