diff --git a/libguh-core/awsconnector.cpp b/libguh-core/awsconnector.cpp index 9db6b93a..b7a0b095 100644 --- a/libguh-core/awsconnector.cpp +++ b/libguh-core/awsconnector.cpp @@ -39,13 +39,17 @@ void AWSConnector::connect2AWS(const QString &endpoint, const QString &clientId, m_client->SetAutoReconnectEnabled(true); m_clientId = clientId; + // subscribe to pairing api topics + subscribe({QString("%1/pair/response").arg(m_clientId), + QString("%1/pair/list/response").arg(m_clientId) + }); + m_connectingFuture = QtConcurrent::run([&]() { ResponseCode rc = m_client->Connect(std::chrono::milliseconds(30000), true, mqtt::Version::MQTT_3_1_1, std::chrono::seconds(60), Utf8String::Create(m_clientId.toStdString()), nullptr, nullptr, nullptr); if (rc == ResponseCode::MQTT_CONNACK_CONNECTION_ACCEPTED) { - qCDebug(dcCloud) << "Connected to AWS."; emit connected(); } else { - qCWarning(dcCloud) << "Error connecting to AWS. Response code:" << QString::fromStdString(ResponseHelper::ToString(rc)); + qCWarning(dcAWS) << "Error connecting to AWS. Response code:" << QString::fromStdString(ResponseHelper::ToString(rc)); m_client.reset(); m_networkConnection.reset(); } @@ -66,19 +70,36 @@ bool AWSConnector::isConnected() const return m_connectingFuture.isFinished() && m_networkConnection && m_client && m_client->IsConnected(); } +void AWSConnector::pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoUserId) +{ + QVariantMap map; + map.insert("idToken", idToken); + map.insert("authToken", authToken); + map.insert("cognitoUserId", cognitoUserId); + map.insert("id", ++m_transactionId); + map.insert("timestamp", QDateTime::currentMSecsSinceEpoch()); + publish(QString("%1/pair").arg(m_clientId), map); + m_pairingRequests.insert(m_transactionId, cognitoUserId); +} + +void AWSConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map) +{ + publish(sessionId + "/reply", map); +} + quint16 AWSConnector::publish(const QString &topic, const QVariantMap &message) { if (!isConnected()) { - qCWarning(dcCloud()) << "Can't publish to AWS: Not connected."; + qCWarning(dcAWS()) << "Can't publish to AWS: Not connected."; return -1; } - QString fullTopic = QString("%1/%2").arg(m_clientId, topic); + QString fullTopic = topic; QJsonDocument jsonDoc = QJsonDocument::fromVariant(message); uint16_t packetId = 0; ResponseCode res = m_client->PublishAsync(Utf8String::Create(fullTopic.toStdString()), false, false, mqtt::QoS::QOS1, jsonDoc.toJson().toStdString(), &publishCallback, packetId); - qCDebug(dcCloud()) << "publish call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId; + qCDebug(dcAWS()) << "publish call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId << "for topic" << topic << jsonDoc.toJson(); s_requestMap.insert(packetId, this); return packetId; } @@ -88,33 +109,42 @@ void AWSConnector::subscribe(const QStringList &topics) m_subscribedTopics.append(topics); if (!isConnected()) { - qCDebug(dcCloud()) << "Can't subscribe to AWS: Not connected. Subscription will happen upon next connection."; + qCDebug(dcAWS()) << "Can't subscribe to AWS: Not connected. Subscription will happen upon next connection."; return; } - subscribeInternally(topics); + doSubscribe(topics); } void AWSConnector::onConnected() { + qCDebug(dcAWS()) << "AWS connected"; if (!m_subscribedTopics.isEmpty()) { - subscribeInternally(m_subscribedTopics); + doSubscribe(m_subscribedTopics); } + retrievePairedDeviceInfo(); } -void AWSConnector::subscribeInternally(const QStringList &topics) +void AWSConnector::retrievePairedDeviceInfo() +{ + QVariantMap params; + params.insert("timestamp", QDateTime::currentMSecsSinceEpoch()); + params.insert("id", ++m_transactionId); + publish(QString("%1/pair/list").arg(m_clientId), params); +} + +void AWSConnector::doSubscribe(const QStringList &topics) { util::Vector> subscription_list; foreach (const QString &topic, topics) { - QString finalTopic = QString("%1/%2").arg(m_clientId, topic); - qCDebug(dcCloud()) << "topic to subscribe is" << finalTopic << "is valid topic:" << Subscription::IsValidTopicName(finalTopic.toStdString()); - auto subscription = mqtt::Subscription::Create(Utf8String::Create(finalTopic.toStdString()), mqtt::QoS::QOS1, &onSubscriptionReceivedCallback, std::shared_ptr(this)); + qCDebug(dcAWS()) << "topic to subscribe is" << topic << "is valid topic:" << Subscription::IsValidTopicName(topic.toStdString()); + auto subscription = mqtt::Subscription::Create(Utf8String::Create(topic.toStdString()), mqtt::QoS::QOS1, &onSubscriptionReceivedCallback, std::shared_ptr(this)); subscription_list.push_back(subscription); } uint16_t packetId; ResponseCode res = m_client->SubscribeAsync(subscription_list, subscribeCallback, packetId); - qCDebug(dcCloud()) << "subscribe call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId; + qCDebug(dcAWS()) << "subscribe call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId; s_requestMap.insert(packetId, this); } @@ -122,24 +152,22 @@ void AWSConnector::publishCallback(uint16_t actionId, ResponseCode rc) { AWSConnector* obj = s_requestMap.take(actionId); if (!obj) { - qCWarning(dcCloud())<< "Received a response callback but don't have an object waiting for it."; + qCWarning(dcAWS())<< "Received a response callback but don't have an object waiting for it."; return; } switch (rc) { case ResponseCode::SUCCESS: - emit obj->responseReceived(actionId, true); - qCDebug(dcCloud()) << "Successfully published" << actionId; + qCDebug(dcAWS()) << "Successfully published" << actionId; break; default: - qCDebug(dcCloud())<< "Error publishing data to AWS:" << QString::fromStdString(ResponseHelper::ToString(rc)); - emit obj->responseReceived(actionId, false); + qCDebug(dcAWS())<< "Error publishing data to AWS:" << QString::fromStdString(ResponseHelper::ToString(rc)); } } void AWSConnector::subscribeCallback(uint16_t actionId, ResponseCode rc) { - qCDebug(dcCloud()) << "subscribed to topic" << actionId << QString::fromStdString(ResponseHelper::ToString(rc)); + qCDebug(dcAWS()) << "subscribed to topic" << actionId << QString::fromStdString(ResponseHelper::ToString(rc)); } ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_name, util::String payload, std::shared_ptr p_app_handler_data) @@ -147,20 +175,56 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam QJsonParseError error; QJsonDocument jsonDoc = QJsonDocument::fromJson(QByteArray::fromStdString(payload), &error); if (error.error != QJsonParseError::NoError) { - qCDebug(dcCloud()) << "Failed to parse JSON from AWS subscription on topic" << QString::fromStdString(topic_name) << ":" << error.errorString() << "\n" << QString::fromStdString(payload); + qCDebug(dcAWS()) << "Failed to parse JSON from AWS subscription on topic" << QString::fromStdString(topic_name) << ":" << error.errorString() << "\n" << QString::fromStdString(payload); return ResponseCode::JSON_PARSING_ERROR; } AWSConnector *connector = dynamic_cast(p_app_handler_data.get()); QString topic = QString::fromStdString(topic_name); - topic.remove(QRegExp("^" + connector->m_clientId + "/")); - emit connector->subscriptionReceived(topic, jsonDoc.toVariant().toMap()); + if (topic == QString("%1/pair/response").arg(connector->m_clientId)) { + int statusCode = jsonDoc.toVariant().toMap().value("status").toInt(); + int id = jsonDoc.toVariant().toMap().value("id").toInt(); + QString cognitoUserId = connector->m_pairingRequests.take(id); + if (!cognitoUserId.isEmpty()) { + qCDebug(dcAWS()) << "Pairing response for id:" << cognitoUserId << statusCode; + emit connector->devicePaired(cognitoUserId, statusCode); + qCDebug(dcAWS()) << "subbbbing"; + connector->subscribe({QString("eu-west-1:%1/listeningPeer/#").arg(cognitoUserId)}); + } else { + qCWarning(dcAWS()) << "Received a pairing response for a transaction we didn't start"; + } + } else if (topic == QString("%1/pair/list/response").arg(connector->m_clientId)) { + qCDebug(dcAWS) << "have device pairings:" << jsonDoc.toVariant().toMap().value("pairings").toList(); + QStringList topics; + foreach (const QVariant &cognitoId, jsonDoc.toVariant().toMap().value("pairings").toList()) { + topics << QString("eu-west-1:%1/listeningPeer/#").arg(cognitoId.toString()); + } + connector->subscribe(topics); + } else if (topic.contains("listeningPeer") && !topic.contains("reply")) { + static QStringList dupes; + QString id = jsonDoc.toVariant().toMap().value("id").toString(); + QString type = jsonDoc.toVariant().toMap().value("type").toString(); + if (dupes.contains(id+type)) { + qCDebug(dcAWS()) << "Dropping duplicate packet"; + return ResponseCode::SUCCESS; + } + dupes.append(id+type); + +// if (type != "offer") return ResponseCode::SUCCESS; + + qCDebug(dcAWS) << "received webrtc handshake message" << topic << jsonDoc.toJson(); + connector->webRtcHandshakeMessageReceived(topic, jsonDoc.toVariant().toMap()); + } else if (topic.contains("listeningPeer") && 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); + } return ResponseCode::SUCCESS; } ResponseCode AWSConnector::onDisconnected(util::String mqtt_client_id, std::shared_ptr p_app_handler_data) { Q_UNUSED(p_app_handler_data) - qCDebug(dcCloud()) << "disconnected" << QString::fromStdString(mqtt_client_id); + qCDebug(dcAWS()) << "disconnected" << QString::fromStdString(mqtt_client_id); return ResponseCode::SUCCESS; } diff --git a/libguh-core/awsconnector.h b/libguh-core/awsconnector.h index 7052e25d..9747830d 100644 --- a/libguh-core/awsconnector.h +++ b/libguh-core/awsconnector.h @@ -19,20 +19,23 @@ public: void disconnectAWS(); bool isConnected() const; - quint16 publish(const QString &topic, const QVariantMap &message); + void pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoUserId); - void subscribe(const QStringList &topics); + void sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map); signals: void connected(); - void responseReceived(quint16 id, bool success); - void subscriptionReceived(const QString &topic, const QVariantMap &data); + void devicePaired(const QString &cognritoUserId, int errorCode); + void webRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); private slots: void onConnected(); + void retrievePairedDeviceInfo(); private: - void subscribeInternally(const QStringList &topcis); + quint16 publish(const QString &topic, const QVariantMap &message); + void subscribe(const QStringList &topics); + void doSubscribe(const QStringList &topcis); static void publishCallback(uint16_t actionId, awsiotsdk::ResponseCode rc); static void subscribeCallback(uint16_t actionId, awsiotsdk::ResponseCode rc); static awsiotsdk::ResponseCode onSubscriptionReceivedCallback(awsiotsdk::util::String topic_name, awsiotsdk::util::String payload, @@ -48,6 +51,9 @@ private: QFuture m_connectingFuture; QStringList m_subscribedTopics; + int m_transactionId = 0; + QHash m_pairingRequests; + static QHash s_requestMap; }; diff --git a/libguh-core/cloudmanager.cpp b/libguh-core/cloudmanager.cpp index 963782b1..07db8fd3 100644 --- a/libguh-core/cloudmanager.cpp +++ b/libguh-core/cloudmanager.cpp @@ -7,7 +7,8 @@ CloudManager::CloudManager(QObject *parent) : QObject(parent) { m_awsConnector = new AWSConnector(this); - connect(m_awsConnector, &AWSConnector::subscriptionReceived, this, &CloudManager::subscriptionReceived); + connect(m_awsConnector, &AWSConnector::devicePaired, this, &CloudManager::onPairingFinished); + connect(m_awsConnector, &AWSConnector::webRtcHandshakeMessageReceived, this, &CloudManager::onAWSWebRtcHandshakeMessageReceived); // Extract the machine id so we have a unique identifier for this machine // TODO: this only works for debian based systems, perhaps we should find something more general @@ -16,11 +17,13 @@ CloudManager::CloudManager(QObject *parent) : QObject(parent) m_deviceId = QString::fromLatin1(f.readAll()).trimmed(); qCDebug(dcCloud()) << "Device ID is:" << m_deviceId; setEnabled(true); - m_awsConnector->subscribe({"pair/response"}); } else { qWarning(dcCloud()) << "Failed to open /etc/machine-id for reading. Cloud connection will not work."; } + m_janusConnector = new JanusConnector(this); + connect(m_janusConnector, &JanusConnector::webRtcHandshakeMessageReceived, this, &CloudManager::onJanusWebRtcHandshakeMessageReceived); + connect(GuhCore::instance()->networkManager(), &NetworkManager::stateChanged, this, &CloudManager::onlineStateChanged); } @@ -83,21 +86,16 @@ void CloudManager::setEnabled(bool enabled) } } -int CloudManager::pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId) +void CloudManager::pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId) { - QVariantMap map; - map.insert("id", ++m_id); - map.insert("idToken", idToken); - map.insert("authToken", authToken); - map.insert("cognitoId", cognitoId); - m_awsConnector->publish("pair", map); - return m_id; + m_awsConnector->pairDevice(idToken, authToken, cognitoId); } void CloudManager::connect2aws() { m_awsConnector->connect2AWS(GuhCore::instance()->configuration()->cloudServerUrl(), - m_deviceId, + "1e10fb7e-d9d9-4145-88dd-2d3caf623c18", +// m_deviceId, GuhCore::instance()->configuration()->cloudCertificateCA(), GuhCore::instance()->configuration()->cloudCertificate(), GuhCore::instance()->configuration()->cloudCertificateKey() @@ -114,10 +112,17 @@ void CloudManager::onlineStateChanged() } } -void CloudManager::subscriptionReceived(const QString &topic, const QVariantMap &message) +void CloudManager::onPairingFinished(const QString &cognitoUserId, int errorCode) { - qCDebug(dcCloud()) << "subscription reveiv ed" << topic << message; - if (topic == "pair/response") { - emit pairingReply(message.value("id").toInt(), message.value("status").toInt()); - } + emit pairingReply(cognitoUserId, errorCode); +} + +void CloudManager::onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data) +{ + m_janusConnector->sendWebRtcHandshakeMessage(transactionId, data); +} + +void CloudManager::onJanusWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data) +{ + m_awsConnector->sendWebRtcHandshakeMessage(transactionId, data); } diff --git a/libguh-core/cloudmanager.h b/libguh-core/cloudmanager.h index bc5a3043..4289d5ef 100644 --- a/libguh-core/cloudmanager.h +++ b/libguh-core/cloudmanager.h @@ -2,6 +2,7 @@ #define CLOUDMANAGER_H #include "awsconnector.h" +#include "janusconnector.h" #include #include @@ -20,24 +21,26 @@ public: bool enabled() const; void setEnabled(bool enabled); - int pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId); + void pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId); signals: - void pairingReply(int pairingTransactionId, int status); + void pairingReply(QString cognitoUserId, int status); private: void connect2aws(); private slots: void onlineStateChanged(); - void subscriptionReceived(const QString &topic, const QVariantMap &message); + void onPairingFinished(const QString &cognitoUserId, int errorCode); + void onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); + void onJanusWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); private: QNetworkSession *m_networkSession; QTimer m_reconnectTimer; bool m_enabled = false; AWSConnector *m_awsConnector = nullptr; - int m_id = 0; // id for transactions. e.g. pairDevice + JanusConnector *m_janusConnector = nullptr; QString m_serverUrl; QString m_deviceId; diff --git a/libguh-core/guhconfiguration.h b/libguh-core/guhconfiguration.h index 525a09e0..5103b95c 100644 --- a/libguh-core/guhconfiguration.h +++ b/libguh-core/guhconfiguration.h @@ -88,6 +88,7 @@ public: QString sslCertificate() const; QString sslCertificateKey() const; + void setSslCertificate(const QString &certificate, const QString &certificateKey); // TCP server QHash tcpServerConfigurations() const; diff --git a/libguh-core/janusconnector.cpp b/libguh-core/janusconnector.cpp new file mode 100644 index 00000000..92147555 --- /dev/null +++ b/libguh-core/janusconnector.cpp @@ -0,0 +1,362 @@ +#include +#include + +#include "janusconnector.h" +#include "loggingcategories.h" + +#include +#include + +JanusConnector::JanusConnector(QObject *parent) : QObject(parent) +{ + m_socket = new QLocalSocket(this); + typedef void (QLocalSocket:: *errorSignal)(QLocalSocket::LocalSocketError); + 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_socket, &QLocalSocket::stateChanged, [&](QLocalSocket::LocalSocketState state) { + qCWarning(dcJanus()) << "state changed" << state; + }); + connect(m_socket, &QLocalSocket::readChannelFinished, [&]() { + qCWarning(dcJanus()) << "readchannel finished"; + }); + connect(&m_socketTimeoutTimer, &QTimer::timeout, this, &JanusConnector::heartbeat); + m_socketTimeoutTimer.setInterval(5000); +} + +bool JanusConnector::connectToJanus() +{ + + int sock = socket (PF_UNIX, SOCK_SEQPACKET, 0); + if (sock < 0) { + qCWarning(dcJanus) << "Failed to create socket"; + return false; + } + + struct sockaddr_un name; + name.sun_family = AF_UNIX; + strncpy (name.sun_path, "/tmp/janusapi", 13); + name.sun_path[13] = '\0'; + int ret = ::connect(sock, (const sockaddr*)&name, sizeof(name)); + if (ret < 0) { + qCWarning(dcJanus()) << "Error connecting to socket"; + return false; + } + + m_socket->setSocketDescriptor(sock); + + m_socketTimeoutTimer.start(); + + return true; +} + +void JanusConnector::createSession(JanusConnector::WebRtcSession *session) +{ + // Open the session + qCDebug(dcJanus()) << "Creating new janus session:" << session; + QString transactionId = QUuid::createUuid().toString(); + QVariantMap map; + map.insert("transaction", transactionId); + map.insert("janus", "create"); + QJsonDocument jsonDoc = QJsonDocument::fromVariant(map); + m_pendingRequests.insert(transactionId, session); + writeToJanus(jsonDoc.toJson()); +} + +void JanusConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &message) +{ + if (!m_socket->isOpen()) { + if (!connectToJanus()) { + qCWarning(dcJanus()) << "Failed to establish a connection to Janus. Cannot send WebRtcHandshake."; + return; + } + } + QString messageType = message.value("type").toString(); + QString transactionId = message.value("id").toString(); + + // Do we have a session for this transaction? + if (!m_sessions.contains(sessionId)) { + + // create a new session and queue the actual offer call + WebRtcSession *session = new WebRtcSession(); + session->sessionId = sessionId; + if (messageType == "offer") { + session->offer = message; + session->offerSent = false; + } else if (messageType == "trickle") { + session->trickles.append(message); + } else { + qCWarning(dcJanus()) << "Unhandled webrtc handshake message type" << messageType << message; + } + m_sessions.insert(sessionId, session); + + if (messageType == "offer") { + createSession(session); + } + return; + } + + WebRtcSession *session = m_sessions.value(sessionId); + if (messageType == "offer") { + session->offer = message; + session->offerSent = false; + createSession(session); + } else if (messageType == "trickle") { + m_sessions.value(sessionId)->trickles.append(message); + } else if (messageType == "webrtcup") { + // If we got the webrtc up from Janus already, directly reply with an ack + if (session->webRtcConnected) { + QVariantMap ack; + ack.insert("id", message.value("id").toString()); + ack.insert("type", "ack"); +// emit webRtcHandshakeMessageReceived(session->sessionId, ack); + } else { + // otherwise store the request and reply when we get the webrtcup + session->webRtcUp = message; + } + } else { + qCWarning(dcJanus()) << "Unhandled message type:" << messageType << message; + } + + processQueue(); +} + +void JanusConnector::processQueue() +{ + if (!m_socket->isOpen()) { + qCWarning(dcJanus()) << "janus socket not open. Cannot process queue"; + return; + } + + foreach (WebRtcSession* session, m_sessions) { + if (session->connectedToJanus) { + if (!session->offerSent) { + QVariantMap janusMessage; + janusMessage.insert("janus", "message"); + janusMessage.insert("transaction", session->offer.value("id").toString()); + janusMessage.insert("session_id", session->janusSessionId); + janusMessage.insert("handle_id", session->janusChannelId); + QVariantMap body; + body.insert("request", "setup"); + janusMessage.insert("body", body); + janusMessage.insert("jsep", session->offer.value("jsep")); + + m_pendingRequests.insert(session->offer.value("id").toString(), session); + + QJsonDocument jsonDoc = QJsonDocument::fromVariant(janusMessage); + qCDebug(dcJanus()) << "Sending offer message to session" << session << jsonDoc.toJson(); + writeToJanus(jsonDoc.toJson()); + session->offerSent = true; + return; + } + while (!session->trickles.isEmpty()) { + QVariantMap input = session->trickles.takeFirst().toMap(); + QVariantMap janusMessage; + janusMessage.insert("janus", "trickle"); + janusMessage.insert("transaction", input.value("id").toString()); + janusMessage.insert("session_id", session->janusSessionId); + janusMessage.insert("handle_id", session->janusChannelId); + janusMessage.insert("candidate", input.value("candidate")); + + m_pendingRequests.insert(input.value("id").toString(), session); + + QJsonDocument jsonDoc = QJsonDocument::fromVariant(janusMessage); + qCDebug(dcJanus()) << "sending trickle message" << jsonDoc.toJson(); + m_socket->write(jsonDoc.toJson()); + return; + } + } + } +} + +void JanusConnector::onDisconnected() +{ + qCDebug(dcJanus) << "disconnected from Janus"; +} + +void JanusConnector::onError(QLocalSocket::LocalSocketError socketError) +{ + qCDebug(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; + QJsonParseError error; + QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error); + if (error.error != QJsonParseError::NoError) { + qWarning(dcJanus()) << "Cannot parse packet received by Janus:" << error.error << error.errorString(); + return; + } + + QVariantMap map = jsonDoc.toVariant().toMap(); + if (map.value("janus").toString() == "error") { + qCWarning(dcJanus()) << "An error happened in the janus connection:" << map.value("error").toMap().value("reason").toString(); + return; + } + + if (map.value("janus").toString() == "timeout") { + quint64 sessionId = map.value("session_id").toLongLong(); + foreach (WebRtcSession *session, m_sessions) { + if (session->matchJanusSessionId(sessionId)) { + qCDebug(dcJanus()) << "Session" << session << "timed out. Removing session"; + m_sessions.remove(session->sessionId); + delete session; + return; + } + } + qCWarning(dcJanus()) << "Received a timeout event but don't have a session for it"; + return; + } + + if (map.value("janus").toString() == "webrtcup") { + quint64 sessionId = map.value("session_id").toLongLong(); + foreach (WebRtcSession *session, m_sessions) { + if (session->matchJanusSessionId(sessionId)) { + qCDebug(dcJanus()) << "Session" << session << "is up now!"; + session->webRtcConnected = true; + if (!session->webRtcUp.isEmpty()) { + QVariantMap ack; + ack.insert("id", session->webRtcUp.value("id").toString()); + ack.insert("type", "ack"); + emit webRtcHandshakeMessageReceived(session->sessionId, ack); + } + delete session; + return; + } + } + qCWarning(dcJanus()) << "Received a webrtcup event 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; + return; + } + + QString transactionId = map.value("transaction").toString(); + WebRtcSession *session = m_pendingRequests.value(transactionId); + if (!session) { + if (transactionId == "pingety") { + qCDebug(dcJanus()) << "Received PONG from Janus"; + return; + } + qCWarning(dcJanus()) << "received a janus message for a session we don't know..."; + return; + } + + if (session->janusSessionId == -1) { + // This should be a create session response + if (map.value("janus").toString() == "success") { + session->janusSessionId = map.value("data").toMap().value("id").toLongLong(); + // oooohhhkaaay... now, this is gonna be dirty... So, Janus' session id is like a freakin huge number + // so that QVariant stores it in a double instead of a longlong, which could cause rounding errors when converting it + // back to to a long. Let's grep the raw data for the parsed session id and if not found, try to correct it one down. + if (!data.contains(QByteArray::number(session->janusSessionId)) && data.contains(QByteArray::number(session->janusSessionId-1))) { + session->janusSessionId--; + qCDebug(dcJanus()) << "corrected session id after rounding error"; + } + qCDebug(dcJanus()) << "Session" << session << "established";// << data << map; + + createChannel(session); + return; + } + qCWarning(dcJanus()) << "Error establishing session"; + delete m_sessions.take(session->sessionId); + return; + } + + qint64 janusSessionId = map.value("session_id").toLongLong(); + if (!session->matchJanusSessionId(janusSessionId)) { + qCWarning(dcJanus) << "Transaction id and session id not matching!" << session->janusSessionId << "!=" << map.value("session_id").toLongLong(); + return; + } + + if (session->janusChannelId == -1) { + if (map.value("janus").toString() == "success") { + session->janusChannelId = map.value("data").toMap().value("id").toLongLong(); + if (!data.contains(QByteArray::number(session->janusChannelId)) && data.contains(QByteArray::number(session->janusChannelId-1))) { + session->janusChannelId--; + qCDebug(dcJanus()) << "Corrected channel id after rounding error"; + } + qCDebug(dcJanus()) << "Channel for session" << session << "established";// << data << map; + session->connectedToJanus = true; + processQueue(); + return; + } + qCWarning(dcJanus()) << "Error establishing channel" << session << data; + return; + } + + if (map.value("janus").toString() == "event" && map.value("jsep").toMap().value("type").toString() == "answer") { + qCDebug(dcJanus()) << "Emitting handshake event" << data; + QVariantMap reply; + reply.insert("id", transactionId); + reply.insert("type", "answer"); + reply.insert("jsep", map.value("jsep")); + emit webRtcHandshakeMessageReceived(session->sessionId, reply); + return; + } + + if (map.value("janus").toString() == "ack") { + QVariantMap ackReply; + ackReply.insert("id", transactionId); + ackReply.insert("type", "ack"); + emit webRtcHandshakeMessageReceived(session->sessionId, ackReply); + return; + } + + qCWarning(dcJanus()) << "Unhandled Janus message:" << data; +} + +void JanusConnector::heartbeat() +{ + QVariantMap map; + map.insert("janus", "ping"); + map.insert("transaction", "pingety"); + QJsonDocument jsonDoc = QJsonDocument::fromVariant(map); + qCDebug(dcJanus()) << "Sending PING to Janus"; + m_socket->write(jsonDoc.toJson()); + m_socket->flush(); +} + +void JanusConnector::createChannel(WebRtcSession *session) +{ + QVariantMap attachPluginMessage; + attachPluginMessage.insert("janus", "attach"); + attachPluginMessage.insert("session_id", session->janusSessionId); + QString transactionId = QUuid::createUuid().toString(); + m_pendingRequests.insert(transactionId, session); + attachPluginMessage.insert("transaction", transactionId); + attachPluginMessage.insert("plugin", "janus.plugin.guhio"); + attachPluginMessage.insert("opaque_id", "guhio-" + QUuid::createUuid().toString()); + QJsonDocument jsonDoc = QJsonDocument::fromVariant(attachPluginMessage); + qCDebug(dcJanus()) << "Establishing channel for session" << session->sessionId; + writeToJanus(jsonDoc.toJson()); +} + +void JanusConnector::writeToJanus(const QByteArray &data) +{ + qCDebug(dcJanus()) << "Writing to janus" << data; + m_socket->write(data); + m_socket->flush(); +} + +QDebug operator<<(QDebug debug, const JanusConnector::WebRtcSession &session) +{ + debug.nospace() << session.sessionId << " (Janus session: " << session.janusSessionId << " channel: " << session.janusChannelId << " connected to Janus: " << session.connectedToJanus << " WebRTC connected: " << session.webRtcConnected << ") "; + return debug; +} + +QDebug operator<<(QDebug debug, JanusConnector::WebRtcSession *session) +{ + debug.nospace() << *session; + return debug; +} diff --git a/libguh-core/janusconnector.h b/libguh-core/janusconnector.h new file mode 100644 index 00000000..5bf14f72 --- /dev/null +++ b/libguh-core/janusconnector.h @@ -0,0 +1,69 @@ +#ifndef JANUSCONNECTOR_H +#define JANUSCONNECTOR_H + +#include +#include +#include + +class JanusConnector : public QObject +{ + Q_OBJECT +public: + class WebRtcSession { + public: + QString sessionId; // This should be unique but persistent for each remote device + qint64 janusSessionId = -1; // the session id for the janus session. + qint64 janusChannelId = -1; + bool connectedToJanus = false; + + QVariantMap offer; + bool offerSent = false; + QVariantList trickles; + QVariantMap webRtcUp; + bool webRtcConnected = false; + + bool matchJanusSessionId(qint64 id) { + return (janusSessionId == id) || (janusSessionId == id +1) || (janusSessionId == id -1); + } + }; + + explicit JanusConnector(QObject *parent = nullptr); + + + void sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &message); + +signals: + void connected(); + void webRtcHandshakeMessageReceived(const QString &sessionId, const QVariantMap &message); + +private slots: + void onDisconnected(); + void onError(QLocalSocket::LocalSocketError socketError); + void onTextMessageReceived(const QString &message); + void onReadyRead(); + void heartbeat(); + void processQueue(); + +private: + QHash m_pendingRequests; + + bool connectToJanus(); + void createSession(WebRtcSession *session); + void createChannel(WebRtcSession *session); + void writeToJanus(const QByteArray &data); + + +private: + + QLocalSocket *m_socket = nullptr; + QTimer m_socketTimeoutTimer; + + QHash m_sessions; + + QStringList m_wantedAcks; +}; + +QDebug operator<<(QDebug debug, const JanusConnector::WebRtcSession &session); +QDebug operator<<(QDebug debug, JanusConnector::WebRtcSession *session); + +#endif // JANUSCONNECTOR_H diff --git a/libguh-core/jsonrpc/jsonrpcserver.cpp b/libguh-core/jsonrpc/jsonrpcserver.cpp index 51b5e972..5d0bf3b5 100644 --- a/libguh-core/jsonrpc/jsonrpcserver.cpp +++ b/libguh-core/jsonrpc/jsonrpcserver.cpp @@ -263,13 +263,14 @@ JsonReply *JsonRPCServer::RemoveToken(const QVariantMap ¶ms) JsonReply *JsonRPCServer::SetupRemoteAccess(const QVariantMap ¶ms) { - - int id = GuhCore::instance()->cloudManager()->pairDevice(params.value("idToken").toString(), params.value("authToken").toString(), params.value("cognitoId").toString()); - qWarning() << "waiting for id" << id; + QString idToken = params.value("idToken").toString(); + QString authToken = params.value("authToken").toString(); + QString cognitoUserId = params.value("cognitoId").toString(); + GuhCore::instance()->cloudManager()->pairDevice(idToken, authToken, cognitoUserId); JsonReply *reply = createAsyncReply("SetupRemoteAccess"); - m_pairingRequests.insert(id, reply); - connect(reply, &JsonReply::finished, [this, id](){ - m_pairingRequests.remove(id); + m_pairingRequests.insert(cognitoUserId, reply); + connect(reply, &JsonReply::finished, [this, cognitoUserId](){ + m_pairingRequests.remove(cognitoUserId); }); return reply; } @@ -496,14 +497,13 @@ void JsonRPCServer::asyncReplyFinished() reply->deleteLater(); } -void JsonRPCServer::pairingFinished(int pairingTransactionId, int status) +void JsonRPCServer::pairingFinished(QString cognitoUserId, int status) { - qWarning() << "pairing finished" << pairingTransactionId << status; - JsonReply *reply = m_pairingRequests.take(pairingTransactionId); + qWarning() << "pairing finished" << cognitoUserId << status; + JsonReply *reply = m_pairingRequests.take(cognitoUserId); if (!reply) { return; } - qWarning() << "blubbbbb"; QVariantMap returns; returns.insert("status", status); reply->setData(returns); diff --git a/libguh-core/jsonrpc/jsonrpcserver.h b/libguh-core/jsonrpc/jsonrpcserver.h index 99b0d669..5f289e9f 100644 --- a/libguh-core/jsonrpc/jsonrpcserver.h +++ b/libguh-core/jsonrpc/jsonrpcserver.h @@ -81,7 +81,7 @@ private slots: void asyncReplyFinished(); - void pairingFinished(int pairingTransactionId, int status); + void pairingFinished(QString cognitoUserId, int status); private: QMap m_interfaces; @@ -91,7 +91,7 @@ private: // clientId, notificationsEnabled QHash m_clients; - QHash m_pairingRequests; + QHash m_pairingRequests; int m_notificationId; diff --git a/libguh-core/libguh-core.pro b/libguh-core/libguh-core.pro index ac39fe6a..38051042 100644 --- a/libguh-core/libguh-core.pro +++ b/libguh-core/libguh-core.pro @@ -70,7 +70,7 @@ HEADERS += $$top_srcdir/libguh-core/guhcore.h \ $$top_srcdir/libguh-core/awsconnector.h \ $$top_srcdir/libguh-core/cloudconnector.h \ $$top_srcdir/libguh-core/MbedTLS/MbedTLSConnection.hpp \ - + $$top_srcdir/libguh-core/janusconnector.h \ SOURCES += $$top_srcdir/libguh-core/guhcore.cpp \ $$top_srcdir/libguh-core/tcpserver.cpp \ @@ -127,3 +127,4 @@ SOURCES += $$top_srcdir/libguh-core/guhcore.cpp \ $$top_srcdir/libguh-core/awsconnector.cpp \ $$top_srcdir/libguh-core/cloudconnector.cpp \ $$top_srcdir/libguh-core/MbedTLS/MbedTLSConnection.cpp \ + $$top_srcdir/libguh-core/janusconnector.cpp \ diff --git a/libguh-core/websocketserver.cpp b/libguh-core/websocketserver.cpp index e1e4b509..f143d6ad 100644 --- a/libguh-core/websocketserver.cpp +++ b/libguh-core/websocketserver.cpp @@ -91,6 +91,7 @@ void WebSocketServer::sendData(const QUuid &clientId, const QByteArray &data) QWebSocket *client = 0; client = m_clientList.value(clientId); if (client) { + qCDebug(dcWebSocketServer()) << "Sending data to client" << data; client->sendTextMessage(data + '\n'); } } diff --git a/libguh/loggingcategories.cpp b/libguh/loggingcategories.cpp index e2060c60..eea3cf74 100644 --- a/libguh/loggingcategories.cpp +++ b/libguh/loggingcategories.cpp @@ -39,3 +39,5 @@ Q_LOGGING_CATEGORY(dcAvahi, "Avahi") Q_LOGGING_CATEGORY(dcCloud, "Cloud") Q_LOGGING_CATEGORY(dcNetworkManager, "NetworkManager") Q_LOGGING_CATEGORY(dcUserManager, "UserManager") +Q_LOGGING_CATEGORY(dcAWS, "AWS") +Q_LOGGING_CATEGORY(dcJanus, "Janus") diff --git a/libguh/loggingcategories.h b/libguh/loggingcategories.h index c6d66bba..e438bc48 100644 --- a/libguh/loggingcategories.h +++ b/libguh/loggingcategories.h @@ -47,5 +47,7 @@ Q_DECLARE_LOGGING_CATEGORY(dcAvahi) Q_DECLARE_LOGGING_CATEGORY(dcCloud) Q_DECLARE_LOGGING_CATEGORY(dcNetworkManager) Q_DECLARE_LOGGING_CATEGORY(dcUserManager) +Q_DECLARE_LOGGING_CATEGORY(dcAWS) +Q_DECLARE_LOGGING_CATEGORY(dcJanus) #endif // LOGGINGCATEGORYS_H diff --git a/server/main.cpp b/server/main.cpp index b39dae68..4c7a0544 100644 --- a/server/main.cpp +++ b/server/main.cpp @@ -127,6 +127,8 @@ int main(int argc, char *argv[]) s_loggingFilters.insert("Cloud", true); s_loggingFilters.insert("NetworkManager", true); s_loggingFilters.insert("UserManager", true); + s_loggingFilters.insert("AWS", false); + s_loggingFilters.insert("Janus", false); QHash loggingFiltersPlugins; foreach (const QJsonObject &pluginMetadata, DeviceManager::pluginsMetadata()) { diff --git a/translations/guhd-de_DE.ts b/translations/guhd-de_DE.ts index 11d15659..1e7c5362 100644 --- a/translations/guhd-de_DE.ts +++ b/translations/guhd-de_DE.ts @@ -4,7 +4,7 @@ main - + guh ( /[guːh]/ ) is an open source IoT (Internet of Things) server, which allows to control a lot of different devices from many different @@ -23,12 +23,12 @@ Szenen undVerhaltensweisen des Systems festzulegen. - + Run guhd in the foreground, not as daemon. Starte guhd im Vordergrund, nicht als Service. - + Debug categories to enable. Prefix with "No" to disable. Warnings from all categories will be printed unless explicitly muted with "NoWarnings". Categories are: @@ -36,17 +36,17 @@ Categories are: Es gibt folgende Kategorien: - + Enables all debug categories. This parameter overrides all debug category parameters. Aktiviere alle Debug-Kategorien. Dieser Parameter überschreibt alle anderen Debug-Kategorien Parameter. - + Specify a log file to write to, If this option is not specified, logs will be printed to the standard output. - + No such debug category: Diese Debug-Kategorie existiert nicht: diff --git a/translations/guhd-en_US.ts b/translations/guhd-en_US.ts index 1d1175e6..fc03f892 100644 --- a/translations/guhd-en_US.ts +++ b/translations/guhd-en_US.ts @@ -4,7 +4,7 @@ main - + guh ( /[guːh]/ ) is an open source IoT (Internet of Things) server, which allows to control a lot of different devices from many different @@ -23,12 +23,12 @@ for your environment. - + Run guhd in the foreground, not as daemon. Run guhd in the foreground, not as daemon. - + Debug categories to enable. Prefix with "No" to disable. Warnings from all categories will be printed unless explicitly muted with "NoWarnings". Categories are: @@ -37,17 +37,17 @@ Categories are: Categories are: - + Enables all debug categories. This parameter overrides all debug category parameters. Enables all debug categories. This parameter overrides all debug category parameters. - + Specify a log file to write to, If this option is not specified, logs will be printed to the standard output. - + No such debug category: No such debug category: