ETM-Schurig/powersync-core
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

more stabilizing, new api changes added

Browse Source
This commit is contained in:
Michael Zanetti 2017-09-24 17:35:52 +02:00
parent cef689c292
commit a31adc5a18
10 changed files with 69 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
libguh-core/MbedTLS/MbedTLSConnection.hpp
View File

@ -25,7 +25,11 @@
#include "mbedtls/config.h"
#include "mbedtls/platform.h"
#ifdef MBEDTLS_NEW_HEADERS
#include "mbedtls/net_sockets.h"
#else
#include "mbedtls/net.h"
#endif
#include "mbedtls/ssl.h"
#include "mbedtls/entropy.h"
#include "mbedtls/ctr_drbg.h"

34
libguh-core/awsconnector.cpp
View File

@ -40,10 +40,9 @@ void AWSConnector::connect2AWS(const QString &endpoint, const QString &clientId,
m_clientId = clientId;
// subscribe to pairing api topics
subscribe({QString("%1/pair/response").arg(m_clientId),
QString("%1/pair/list/response").arg(m_clientId)
});
subscribe({QString("create/device/%1").arg(m_clientId)});
qCDebug(dcAWS()) << "Connecting to AWS with ID:" << 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) {
@ -118,10 +117,11 @@ void AWSConnector::subscribe(const QStringList &topics)
void AWSConnector::onConnected()
{
qCDebug(dcAWS()) << "AWS connected";
registerDevice();
retrievePairedDeviceInfo();
if (!m_subscribedTopics.isEmpty()) {
doSubscribe(m_subscribedTopics);
}
retrievePairedDeviceInfo();
}
void AWSConnector::retrievePairedDeviceInfo()
@ -132,6 +132,13 @@ void AWSConnector::retrievePairedDeviceInfo()
publish(QString("%1/pair/list").arg(m_clientId), params);
}
void AWSConnector::registerDevice()
{
QVariantMap params;
params.insert("serverUUID", m_clientId);
publish("create/device", params);
}
void AWSConnector::doSubscribe(const QStringList &topics)
{
util::Vector<std::shared_ptr<mqtt::Subscription>> subscription_list;
@ -141,7 +148,6 @@ void AWSConnector::doSubscribe(const QStringList &topics)
subscription_list.push_back(subscription);
}
uint16_t packetId;
ResponseCode res = m_client->SubscribeAsync(subscription_list, subscribeCallback, packetId);
qCDebug(dcAWS()) << "subscribe call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId;
@ -181,14 +187,22 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam
AWSConnector *connector = dynamic_cast<AWSConnector*>(p_app_handler_data.get());
QString topic = QString::fromStdString(topic_name);
if (topic == QString("%1/pair/response").arg(connector->m_clientId)) {
if (topic == QString("create/device/%1").arg(connector->m_clientId)) {
int statusCode = jsonDoc.toVariant().toMap().value("result").toMap().value("code").toInt();
if (statusCode != 200) {
qCWarning(dcAWS()) << "Error registering device in the cloud. AWS connetion will not work.";
return ResponseCode::SUCCESS;
}
qCDebug(dcAWS()) << "Device registered in cloud";
connector->subscribe({QString("%1/pair/response").arg(connector->m_clientId),
QString("%1/pair/list/response").arg(connector->m_clientId)});
} else 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";
@ -196,8 +210,8 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam
} 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());
foreach (const QVariant &pairing, jsonDoc.toVariant().toMap().value("pairings").toList()) {
topics << QString("eu-west-1:%1/listeningPeer/#").arg(pairing.toMap().value("cognitoIdIdentityId").toString());
}
connector->subscribe(topics);
} else if (topic.contains("listeningPeer") && !topic.contains("reply")) {
@ -210,8 +224,6 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam
}
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")) {

1
libguh-core/awsconnector.h
View File

@ -31,6 +31,7 @@ signals:
private slots:
void onConnected();
void retrievePairedDeviceInfo();
void registerDevice();
private:
quint16 publish(const QString &topic, const QVariantMap &message);

11
libguh-core/cloudmanager.cpp
View File

@ -14,7 +14,12 @@ CloudManager::CloudManager(QObject *parent) : QObject(parent)
// TODO: this only works for debian based systems, perhaps we should find something more general
QFile f("/etc/machine-id");
if (f.open(QFile::ReadOnly)) {
m_deviceId = QString::fromLatin1(f.readAll()).trimmed();
QString tmpId = QString::fromLatin1(f.readAll()).trimmed();
tmpId.insert(8, "-");
tmpId.insert(13, "-");
tmpId.insert(18, "-");
tmpId.insert(23, "-");
m_deviceId = QUuid(tmpId);
qCDebug(dcCloud()) << "Device ID is:" << m_deviceId;
setEnabled(true);
} else {
@ -53,7 +58,7 @@ void CloudManager::setEnabled(bool enabled)
{
if (enabled) {
bool missingConfig = false;
if (m_deviceId.isEmpty()) {
if (m_deviceId.isNull()) {
qCWarning(dcCloud()) << "Don't have a unique device ID (/etc/machine-id).";
missingConfig = true;
}
@ -95,7 +100,7 @@ void CloudManager::connect2aws()
{
m_awsConnector->connect2AWS(GuhCore::instance()->configuration()->cloudServerUrl(),
"1e10fb7e-d9d9-4145-88dd-2d3caf623c18",
// m_deviceId,
// m_deviceId.toString().remove(QRegExp("[{}]*")),
GuhCore::instance()->configuration()->cloudCertificateCA(),
GuhCore::instance()->configuration()->cloudCertificate(),
GuhCore::instance()->configuration()->cloudCertificateKey()

3
libguh-core/cloudmanager.h
View File

@ -7,6 +7,7 @@
#include <QObject>
#include <QTimer>
#include <QNetworkSession>
#include <QUuid>
class CloudManager : public QObject
{
@ -43,7 +44,7 @@ private:
JanusConnector *m_janusConnector = nullptr;
QString m_serverUrl;
QString m_deviceId;
QUuid m_deviceId;
QString m_caCertificate;
QString m_clientCertificate;
QString m_clientCertificateKey;

9
libguh-core/guhconfiguration.cpp
View File

@ -347,6 +347,15 @@ QString GuhConfiguration::sslCertificateKey() const
return settings.value("certificate-key").toString();
}
void GuhConfiguration::setSslCertificate(const QString &sslCertificate, const QString &sslCertificateKey)
{
GuhSettings settings(GuhSettings::SettingsRoleGlobal);
settings.beginGroup("SSL");
settings.setValue("certificate", sslCertificate);
settings.setValue("certificate-key", sslCertificateKey);
settings.endGroup();
}
void GuhConfiguration::setServerUuid(const QUuid &uuid)
{
qCDebug(dcApplication()) << "Configuration: Server uuid:" << uuid.toString();

2
libguh-core/guhconfiguration.h
View File

@ -88,7 +88,7 @@ public:
QString sslCertificate() const;
QString sslCertificateKey() const;
void setSslCertificate(const QString &certificate, const QString &certificateKey);
void setSslCertificate(const QString &sslCertificate, const QString &sslCertificateKey);
// TCP server
QHash<QString, ServerConfiguration> tcpServerConfigurations() const;

27
libguh-core/janusconnector.cpp
View File

@ -105,15 +105,15 @@ void JanusConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const
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;
// }
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;
}
@ -187,7 +187,7 @@ void JanusConnector::onTextMessageReceived(const QString &message)
void JanusConnector::onReadyRead()
{
QByteArray data = m_socket->readAll();
qCDebug(dcJanus()) << "incoming data" << data;
// qCDebug(dcJanus()) << "incoming data" << data;
QJsonParseError error;
QJsonDocument jsonDoc = QJsonDocument::fromJson(data, &error);
if (error.error != QJsonParseError::NoError) {
@ -211,7 +211,7 @@ void JanusConnector::onReadyRead()
return;
}
}
qCWarning(dcJanus()) << "Received a timeout event but don't have a session for it";
qCWarning(dcJanus()) << "Received a timeout event but don't have a session for it." << data << map.value("session_id").toLongLong();
return;
}
@ -227,7 +227,6 @@ void JanusConnector::onReadyRead()
ack.insert("type", "ack");
emit webRtcHandshakeMessageReceived(session->sessionId, ack);
}
delete session;
return;
}
}
@ -245,7 +244,7 @@ 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;
}
qCWarning(dcJanus()) << "received a janus message for a session we don't know...";
@ -322,7 +321,7 @@ void JanusConnector::heartbeat()
map.insert("janus", "ping");
map.insert("transaction", "pingety");
QJsonDocument jsonDoc = QJsonDocument::fromVariant(map);
qCDebug(dcJanus()) << "Sending PING to Janus";
// qCDebug(dcJanus()) << "Sending PING to Janus";
m_socket->write(jsonDoc.toJson());
m_socket->flush();
}

4
libguh-core/libguh-core.pro
View File

@ -10,6 +10,10 @@ LIBS += -L$$top_builddir/libguh/ -lguh -lssl -lcrypto
target.path = /usr/lib/$$system('dpkg-architecture -q DEB_HOST_MULTIARCH')
INSTALLS += target
exists("/usr/include/mbedtls/net_sockets.h") {
DEFINES += MBEDTLS_NEW_HEADERS
}
# icons for the webserver
RESOURCES += $$top_srcdir/icons.qrc

4
tests/auto/jsonrpc/testjsonrpc.cpp
View File

@ -114,6 +114,10 @@ void TestJSONRPC::testHandshake()
QCOMPARE(handShake.value("params").toMap().value("version").toString(), guhVersionString);
m_mockTcpServer->clientDisconnected(newClientId);
// And now check if it is sent again when calling JSONRPC.Hello
handShake = injectAndWait("JSONRPC.Hello").toMap();
QCOMPARE(handShake.value("params").toMap().value("version").toString(), guhVersionString);
}
void TestJSONRPC::testInitialSetup()