add keepalive api

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);

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,

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);
 

libguh-core/janusconnector.cpp

@@ -34,8 +34,6 @@ JanusConnector::JanusConnector(QObject *parent) : QObject(parent)
     connect(m_socket, static_cast<errorSignal>(&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();
 }
 

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<QString, WebRtcSession*> m_sessions;
 

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 &params)
     return reply;
 }
 
+JsonReply *JsonRPCServer::KeepAlive(const QVariantMap &params)
+{
+    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<QString, JsonHandler *> JsonRPCServer::handlers() const
 {

libguh-core/jsonrpc/jsonrpcserver.h

@@ -57,6 +57,7 @@ public:
     Q_INVOKABLE JsonReply *Tokens(const QVariantMap &params) const;
     Q_INVOKABLE JsonReply *RemoveToken(const QVariantMap &params);
     Q_INVOKABLE JsonReply *SetupRemoteAccess(const QVariantMap &params);
+    Q_INVOKABLE JsonReply *KeepAlive(const QVariantMap &params);
 
     QHash<QString, JsonHandler *> handlers() const;
 

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<QString, bool> loggingFiltersPlugins;