diff --git a/libguh-core/awsconnector.cpp b/libguh-core/awsconnector.cpp index af8e1fba..a2e29969 100644 --- a/libguh-core/awsconnector.cpp +++ b/libguh-core/awsconnector.cpp @@ -392,7 +392,7 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam qCDebug(dcAWS) << "received webrtc handshake message" << topic << jsonDoc.toJson(); connector->webRtcHandshakeMessageReceived(topic, jsonDoc.toVariant().toMap()); - } else if (topic.contains("listeningPeer") && topic.contains("reply")) { + } else if (topic.startsWith(QString("%1/eu-west-1:").arg(connector->m_clientId)) && topic.contains("reply")) { // silently drop our own things (should not be subscribed to that in the first place) } else { qCWarning(dcAWS) << "Unhandled subscription received!" << topic << QString::fromStdString(payload); diff --git a/libguh-core/cloudmanager.cpp b/libguh-core/cloudmanager.cpp index 22106c73..c8217706 100644 --- a/libguh-core/cloudmanager.cpp +++ b/libguh-core/cloudmanager.cpp @@ -116,6 +116,11 @@ void CloudManager::pairDevice(const QString &idToken, const QString &userId) m_awsConnector->pairDevice(idToken, userId); } +bool CloudManager::keepAlive(const QString &sessionId) +{ + return m_janusConnector->sendKeepAliveMessage(sessionId); +} + void CloudManager::connect2aws() { m_awsConnector->connect2AWS(m_serverUrl, diff --git a/libguh-core/cloudmanager.h b/libguh-core/cloudmanager.h index 2a32a12f..5244d2c5 100644 --- a/libguh-core/cloudmanager.h +++ b/libguh-core/cloudmanager.h @@ -48,6 +48,8 @@ public: void pairDevice(const QString &idToken, const QString &userId); + bool keepAlive(const QString &sessionId); + signals: void pairingReply(QString cognitoUserId, int status, const QString &message); diff --git a/libguh-core/janusconnector.cpp b/libguh-core/janusconnector.cpp index f952a02f..6722df78 100644 --- a/libguh-core/janusconnector.cpp +++ b/libguh-core/janusconnector.cpp @@ -34,8 +34,6 @@ JanusConnector::JanusConnector(QObject *parent) : QObject(parent) connect(m_socket, static_cast(&QLocalSocket::error), this, &JanusConnector::onError); connect(m_socket, &QLocalSocket::disconnected, this, &JanusConnector::onDisconnected); connect(m_socket, &QLocalSocket::readyRead, this, &JanusConnector::onReadyRead); - connect(&m_socketTimeoutTimer, &QTimer::timeout, this, &JanusConnector::heartbeat); - m_socketTimeoutTimer.setInterval(5000); connectToJanus(); } @@ -61,8 +59,6 @@ bool JanusConnector::connectToJanus() m_socket->setSocketDescriptor(sock); - m_socketTimeoutTimer.start(); - return true; } @@ -130,6 +126,12 @@ void JanusConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const // otherwise store the request and reply when we get the webrtcup session->webRtcUp = message; } + } else if (messageType == "ack") { + QVariantMap janusMessage; + janusMessage.insert("janus", "ack"); + janusMessage.insert("id", transactionId); + janusMessage.insert("transaction", "ack"); + writeToJanus(QJsonDocument::fromVariant(janusMessage).toJson(QJsonDocument::Compact)); } else { qCWarning(dcJanus()) << "Unhandled message type:" << messageType << message; } @@ -137,6 +139,22 @@ void JanusConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const processQueue(); } +bool JanusConnector::sendKeepAliveMessage(const QString &sessionId) +{ + WebRtcSession *session = m_sessions.value(sessionId); + if (!session) { + qCWarning(dcJanus()) << "Received a keepalive message for a session we don't know."; + return false; + } + QVariantMap janusMessage; + janusMessage.insert("janus", "keepalive"); + janusMessage.insert("session_id", session->janusSessionId); + janusMessage.insert("handle_id", session->janusChannelId); + janusMessage.insert("transaction", "keepalive"); + writeToJanus(QJsonDocument::fromVariant(janusMessage).toJson(QJsonDocument::Compact)); + return true; +} + void JanusConnector::processQueue() { if (!m_socket->isOpen()) { @@ -187,7 +205,7 @@ void JanusConnector::processQueue() void JanusConnector::onDisconnected() { - qCDebug(dcJanus) << "Disconnected from Janus"; + qCDebug(dcJanus) << "Disconnected from Janus" << m_socket->isOpen(); } void JanusConnector::onError(QLocalSocket::LocalSocketError socketError) @@ -195,15 +213,10 @@ void JanusConnector::onError(QLocalSocket::LocalSocketError socketError) qCWarning(dcJanus) << "Error in janus connection" << socketError << m_socket->errorString(); } -void JanusConnector::onTextMessageReceived(const QString &message) -{ - qCDebug(dcJanus) << "Text message received from Janus" << message; -} - void JanusConnector::onReadyRead() { QByteArray data = m_socket->readAll(); -// qCDebug(dcJanus()) << "incoming data" << data; + qCDebug(dcJanusTraffic()) << "incoming data" << data; QJsonParseError error; QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error); if (error.error != QJsonParseError::NoError) { @@ -250,6 +263,23 @@ void JanusConnector::onReadyRead() return; } + if (map.value("janus").toString() == "hangup") { + quint64 sessionId = map.value("session_id").toLongLong(); + foreach (WebRtcSession *session, m_sessions) { + if (session->matchJanusSessionId(sessionId)){ + qCDebug(dcJanus()) << "Session" << session << "hangup received. Reason:" << map.value("reason").toString(); + QVariantMap hangup; + hangup.insert("type", "hangup"); + hangup.insert("reason", map.value("reason").toString()); + emit webRtcHandshakeMessageReceived(session->sessionId, hangup); + m_sessions.remove(session->sessionId); + return; + } + } + qCWarning(dcJanus()) << "Received a hangup message but don't have a session for it"; + return; + } + // as of now, everything must be part of a transaction if (!map.contains("transaction")) { qCWarning(dcJanus) << "Unhandled message from Janus (missing transaction):" << data; @@ -260,7 +290,11 @@ void JanusConnector::onReadyRead() WebRtcSession *session = m_pendingRequests.value(transactionId); if (!session) { if (transactionId == "pingety") { -// qCDebug(dcJanus()) << "Received PONG from Janus"; + qCDebug(dcJanus()) << "Received PONG from Janus"; + return; + } + if (transactionId == "keepalive") { + qCDebug(dcJanus()) << "Keep alive acked by janus."; return; } qCWarning(dcJanus()) << "received a janus message for a session we don't know..."; @@ -359,8 +393,11 @@ void JanusConnector::createChannel(WebRtcSession *session) void JanusConnector::writeToJanus(const QByteArray &data) { - qCDebug(dcJanus()) << "Writing to janus" << data; - m_socket->write(data); + qCDebug(dcJanusTraffic()) << "Writing to janus" << data; + qint64 count = m_socket->write(data); + if (count != data.length()) { + qCWarning(dcJanus()) << "Error writing to Janus."; + } m_socket->flush(); } diff --git a/libguh-core/janusconnector.h b/libguh-core/janusconnector.h index 6c37a69c..ebafbf11 100644 --- a/libguh-core/janusconnector.h +++ b/libguh-core/janusconnector.h @@ -51,6 +51,7 @@ public: void sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &message); + bool sendKeepAliveMessage(const QString &sessionId); signals: void connected(); @@ -59,7 +60,6 @@ signals: private slots: void onDisconnected(); void onError(QLocalSocket::LocalSocketError socketError); - void onTextMessageReceived(const QString &message); void onReadyRead(); void heartbeat(); void processQueue(); @@ -76,7 +76,6 @@ private: private: QLocalSocket *m_socket = nullptr; - QTimer m_socketTimeoutTimer; QHash m_sessions; diff --git a/libguh-core/jsonrpc/jsonrpcserver.cpp b/libguh-core/jsonrpc/jsonrpcserver.cpp index 2bfa33f2..83b133c3 100644 --- a/libguh-core/jsonrpc/jsonrpcserver.cpp +++ b/libguh-core/jsonrpc/jsonrpcserver.cpp @@ -149,6 +149,13 @@ JsonRPCServer::JsonRPCServer(const QSslConfiguration &sslConfiguration, QObject returns.insert("message", JsonTypes::basicTypeToString(JsonTypes::String)); setReturns("SetupRemoteAccess", returns); + params.clear(); returns.clear(); + setDescription("KeepAlive", "Keep alive a remote connection. The sessionId is the MQTT topic which has been used to establish the session. It will return false if no ongoing session with the given ID can be found."); + params.insert("sessionId", JsonTypes::basicTypeToString(JsonTypes::String)); + setParams("KeepAlive", params); + returns.insert("success", JsonTypes::basicTypeToString(JsonTypes::Bool)); + setReturns("KeepAlive", returns); + QMetaObject::invokeMethod(this, "setup", Qt::QueuedConnection); } @@ -274,6 +281,15 @@ JsonReply *JsonRPCServer::SetupRemoteAccess(const QVariantMap ¶ms) return reply; } +JsonReply *JsonRPCServer::KeepAlive(const QVariantMap ¶ms) +{ + QString sessionId = params.value("sessionId").toString(); + bool result = GuhCore::instance()->cloudManager()->keepAlive(sessionId); + QVariantMap resultMap; + resultMap.insert("success", result); + return createReply(resultMap); +} + /*! Returns the list of registred \l{JsonHandler}{JsonHandlers} and their name.*/ QHash JsonRPCServer::handlers() const { diff --git a/libguh-core/jsonrpc/jsonrpcserver.h b/libguh-core/jsonrpc/jsonrpcserver.h index 3bded0a5..8ec95ef3 100644 --- a/libguh-core/jsonrpc/jsonrpcserver.h +++ b/libguh-core/jsonrpc/jsonrpcserver.h @@ -57,6 +57,7 @@ public: Q_INVOKABLE JsonReply *Tokens(const QVariantMap ¶ms) const; Q_INVOKABLE JsonReply *RemoveToken(const QVariantMap ¶ms); Q_INVOKABLE JsonReply *SetupRemoteAccess(const QVariantMap ¶ms); + Q_INVOKABLE JsonReply *KeepAlive(const QVariantMap ¶ms); QHash handlers() const; diff --git a/server/main.cpp b/server/main.cpp index 007ce014..0ef69274 100644 --- a/server/main.cpp +++ b/server/main.cpp @@ -122,6 +122,7 @@ int main(int argc, char *argv[]) s_loggingFilters.insert("WebSocketServer", false); s_loggingFilters.insert("WebSocketServerTraffic", false); s_loggingFilters.insert("JsonRpc", false); + s_loggingFilters.insert("JsonRpcTraffic", false); s_loggingFilters.insert("Rest", false); s_loggingFilters.insert("OAuth2", false); s_loggingFilters.insert("TimeManager", false); @@ -132,6 +133,7 @@ int main(int argc, char *argv[]) s_loggingFilters.insert("UserManager", true); s_loggingFilters.insert("AWS", false); s_loggingFilters.insert("AWSTraffic", false); + s_loggingFilters.insert("Janus", false); s_loggingFilters.insert("JanusTraffic", false); QHash loggingFiltersPlugins;