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

more work after testing with Luca

Browse Source
This commit is contained in:
Michael Zanetti 2017-10-26 18:32:59 +02:00
parent a0371a0133
commit 403a78d029
8 changed files with 48 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
libguh-core/awsconnector.cpp
View File

@ -176,7 +176,7 @@ void AWSConnector::onPairingsRetrieved(const QVariantList &pairings)
{ {
QStringList topics; QStringList topics;
foreach (const QVariant &pairing, pairings) { foreach (const QVariant &pairing, pairings) {
topics << QString("eu-west-1:%1/%2/#").arg(pairing.toMap().value("cognitoIdIdentityId").toString()).arg(m_clientId); topics << QString("%1/%2/#").arg(m_clientId).arg(pairing.toString());
} }
subscribe(topics); subscribe(topics);
@ -195,16 +195,15 @@ bool AWSConnector::isConnected() const
return m_connectingFuture.isFinished() && m_networkConnection && m_client && m_client->IsConnected(); return m_connectingFuture.isFinished() && m_networkConnection && m_client && m_client->IsConnected();
} }
void AWSConnector::pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoUserId) void AWSConnector::pairDevice(const QString &idToken, const QString &userId)
{ {
QVariantMap map; QVariantMap map;
map.insert("idToken", idToken); map.insert("idToken", idToken);
map.insert("authToken", authToken); map.insert("userId", userId);
map.insert("cognitoUserId", cognitoUserId);
map.insert("id", ++m_transactionId); map.insert("id", ++m_transactionId);
map.insert("timestamp", QDateTime::currentMSecsSinceEpoch()); map.insert("timestamp", QDateTime::currentMSecsSinceEpoch());
publish(QString("%1/pair").arg(m_clientId), map); publish(QString("%1/pair").arg(m_clientId), map);
m_pairingRequests.insert(m_transactionId, cognitoUserId); m_pairingRequests.insert(m_transactionId, userId);
} }
void AWSConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map) void AWSConnector::sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map)
@ -222,7 +221,7 @@ quint16 AWSConnector::publish(const QString &topic, const QVariantMap &message)
QJsonDocument jsonDoc = QJsonDocument::fromVariant(message); QJsonDocument jsonDoc = QJsonDocument::fromVariant(message);
uint16_t packetId = 0; uint16_t packetId = 0;
ResponseCode res = m_client->PublishAsync(Utf8String::Create(fullTopic.toStdString()), false, false, mqtt::QoS::QOS1, jsonDoc.toJson().toStdString(), &publishCallback, packetId); ResponseCode res = m_client->PublishAsync(Utf8String::Create(fullTopic.toStdString()), false, false, mqtt::QoS::QOS1, jsonDoc.toJson(QJsonDocument::Compact).toStdString(), &publishCallback, packetId);
qCDebug(dcAWSTraffic()) << "publish call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId << "for topic" << topic << jsonDoc.toJson(); qCDebug(dcAWSTraffic()) << "publish call queued with status:" << QString::fromStdString(ResponseHelper::ToString(res)) << packetId << "for topic" << topic << jsonDoc.toJson();
s_requestMap.insert(packetId, this); s_requestMap.insert(packetId, this);
return packetId; return packetId;
@ -342,6 +341,8 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam
return ResponseCode::JSON_PARSING_ERROR; return ResponseCode::JSON_PARSING_ERROR;
} }
qCDebug(dcAWSTraffic()) << "Subscription received: Topic:" << QString::fromStdString(topic_name) << "payload:" << QString::fromStdString(payload);
AWSConnector *connector = dynamic_cast<SubscriptionContext*>(p_app_handler_data.get())->c; AWSConnector *connector = dynamic_cast<SubscriptionContext*>(p_app_handler_data.get())->c;
QString topic = QString::fromStdString(topic_name); QString topic = QString::fromStdString(topic_name);
if (topic.startsWith("create/device/")) { if (topic.startsWith("create/device/")) {
@ -363,21 +364,25 @@ ResponseCode AWSConnector::onSubscriptionReceivedCallback(util::String topic_nam
} else if (topic == QString("%1/pair/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 statusCode = jsonDoc.toVariant().toMap().value("status").toInt();
int id = jsonDoc.toVariant().toMap().value("id").toInt(); int id = jsonDoc.toVariant().toMap().value("id").toInt();
QString cognitoUserId = connector->m_pairingRequests.take(id); QString message = jsonDoc.toVariant().toMap().value("result").toMap().value("message").toString();
if (!cognitoUserId.isEmpty()) { QString userId = connector->m_pairingRequests.take(id);
qCDebug(dcAWS()) << "Pairing response for id:" << cognitoUserId << statusCode; if (statusCode != 200) {
emit connector->devicePaired(cognitoUserId, statusCode); qCWarning(dcAWS()) << "Pairing failed:" << statusCode << message;
connector->subscribe({QString("eu-west-1:%1/listeningPeer/#").arg(cognitoUserId)}); emit connector->devicePaired(userId, statusCode, message);
} else if (!userId.isEmpty()) {
qCDebug(dcAWS()) << "Pairing response for id:" << userId << statusCode;
emit connector->devicePaired(userId, statusCode, message);
connector->subscribe({QString("eu-west-1:%1/listeningPeer/#").arg(userId)});
} else { } else {
qCWarning(dcAWS()) << "Received a pairing response for a transaction we didn't start"; qCWarning(dcAWS()) << "Received a pairing response for a transaction we didn't start";
} }
} else if (topic == QString("%1/device/users/response").arg(connector->m_clientId)) { } else if (topic == QString("%1/device/users/response").arg(connector->m_clientId)) {
qCDebug(dcAWS) << "have device pairings:" << jsonDoc.toVariant().toMap().value("pairings").toList(); if (jsonDoc.toVariant().toMap().value("users").toList().isEmpty()) {
if (jsonDoc.toVariant().toMap().value("pairings").toList().isEmpty()) {
qCDebug(dcAWS()) << "No devices paired yet..."; qCDebug(dcAWS()) << "No devices paired yet...";
return ResponseCode::SUCCESS; return ResponseCode::SUCCESS;
} }
connector->staticMetaObject.invokeMethod(connector, "inPairingsReceived", Qt::QueuedConnection, Q_ARG(QVariantList, jsonDoc.toVariant().toMap().value("pairings").toList())); qCDebug(dcAWS) << jsonDoc.toVariant().toMap().value("users").toList().count() << "devices paired in cloud.";
connector->staticMetaObject.invokeMethod(connector, "onPairingsRetrieved", Qt::QueuedConnection, Q_ARG(QVariantList, jsonDoc.toVariant().toMap().value("users").toList()));
} else if (topic == QString("%1/device/name/response").arg(connector->m_clientId)) { } else if (topic == QString("%1/device/name/response").arg(connector->m_clientId)) {
qCDebug(dcAWS) << "Set device name in cloud with status:" << jsonDoc.toVariant().toMap().value("status").toInt(); qCDebug(dcAWS) << "Set device name in cloud with status:" << jsonDoc.toVariant().toMap().value("status").toInt();
} else if (topic.startsWith("eu-west-1:") && !topic.contains("reply")) { } else if (topic.startsWith("eu-west-1:") && !topic.contains("reply")) {

6
libguh-core/awsconnector.h
View File

@ -43,14 +43,14 @@ public:
void disconnectAWS(); void disconnectAWS();
bool isConnected() const; bool isConnected() const;
void pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoUserId); void pairDevice(const QString &idToken, const QString &userId);
void sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map); void sendWebRtcHandshakeMessage(const QString &sessionId, const QVariantMap &map);
signals: signals:
void connected(); void connected();
void disconnected(); void disconnected();
void devicePaired(const QString &cognritoUserId, int errorCode); void devicePaired(const QString &cognritoUserId, int errorCode, const QString &message);
void webRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); void webRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data);
private slots: private slots:
@ -101,7 +101,7 @@ private:
QFuture<void> m_connectingFuture; QFuture<void> m_connectingFuture;
bool m_isCleanSession = true; bool m_isCleanSession = true;
int m_transactionId = 0; quint8 m_transactionId = 0;
QString m_createDeviceId; QString m_createDeviceId;
int m_createDeviceSubscriptionId = 0; int m_createDeviceSubscriptionId = 0;
QHash<quint16, QString> m_pairingRequests; QHash<quint16, QString> m_pairingRequests;

8
libguh-core/cloudmanager.cpp
View File

@ -111,9 +111,9 @@ void CloudManager::setEnabled(bool enabled)
} }
} }
void CloudManager::pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId) void CloudManager::pairDevice(const QString &idToken, const QString &userId)
{ {
m_awsConnector->pairDevice(idToken, authToken, cognitoId); m_awsConnector->pairDevice(idToken, userId);
} }
void CloudManager::connect2aws() void CloudManager::connect2aws()
@ -137,9 +137,9 @@ void CloudManager::onlineStateChanged()
} }
} }
void CloudManager::onPairingFinished(const QString &cognitoUserId, int errorCode) void CloudManager::onPairingFinished(const QString &cognitoUserId, int errorCode, const QString &message)
{ {
emit pairingReply(cognitoUserId, errorCode); emit pairingReply(cognitoUserId, errorCode, message);
} }
void CloudManager::onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data) void CloudManager::onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data)

6
libguh-core/cloudmanager.h
View File

@ -46,17 +46,17 @@ public:
bool enabled() const; bool enabled() const;
void setEnabled(bool enabled); void setEnabled(bool enabled);
void pairDevice(const QString &idToken, const QString &authToken, const QString &cognitoId); void pairDevice(const QString &idToken, const QString &userId);
signals: signals:
void pairingReply(QString cognitoUserId, int status); void pairingReply(QString cognitoUserId, int status, const QString &message);
private: private:
void connect2aws(); void connect2aws();
private slots: private slots:
void onlineStateChanged(); void onlineStateChanged();
void onPairingFinished(const QString &cognitoUserId, int errorCode); void onPairingFinished(const QString &cognitoUserId, int errorCode, const QString &message);
void onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); void onAWSWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data);
void onJanusWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data); void onJanusWebRtcHandshakeMessageReceived(const QString &transactionId, const QVariantMap &data);

23
libguh-core/jsonrpc/jsonrpcserver.cpp
View File

@ -143,10 +143,10 @@ JsonRPCServer::JsonRPCServer(const QSslConfiguration &sslConfiguration, QObject
params.clear(); returns.clear(); params.clear(); returns.clear();
setDescription("SetupRemoteAccess", "Setup the remote connection by providing AWS token information"); setDescription("SetupRemoteAccess", "Setup the remote connection by providing AWS token information");
params.insert("idToken", JsonTypes::basicTypeToString(JsonTypes::String)); params.insert("idToken", JsonTypes::basicTypeToString(JsonTypes::String));
params.insert("authToken", JsonTypes::basicTypeToString(JsonTypes::String)); params.insert("userId", JsonTypes::basicTypeToString(JsonTypes::String));
params.insert("cognitoId", JsonTypes::basicTypeToString(JsonTypes::String));
setParams("SetupRemoteAccess", params); setParams("SetupRemoteAccess", params);
returns.insert("status", JsonTypes::basicTypeToString(JsonTypes::Int)); returns.insert("status", JsonTypes::basicTypeToString(JsonTypes::Int));
returns.insert("message", JsonTypes::basicTypeToString(JsonTypes::String));
setReturns("SetupRemoteAccess", returns); setReturns("SetupRemoteAccess", returns);
QMetaObject::invokeMethod(this, "setup", Qt::QueuedConnection); QMetaObject::invokeMethod(this, "setup", Qt::QueuedConnection);
@ -263,14 +263,14 @@ JsonReply *JsonRPCServer::RemoveToken(const QVariantMap &params)
JsonReply *JsonRPCServer::SetupRemoteAccess(const QVariantMap &params) JsonReply *JsonRPCServer::SetupRemoteAccess(const QVariantMap &params)
{ {
qWarning() << "SetupRemoteAccess called";
QString idToken = params.value("idToken").toString(); QString idToken = params.value("idToken").toString();
QString authToken = params.value("authToken").toString(); QString userId = params.value("userId").toString();
QString cognitoUserId = params.value("cognitoId").toString(); GuhCore::instance()->cloudManager()->pairDevice(idToken, userId);
GuhCore::instance()->cloudManager()->pairDevice(idToken, authToken, cognitoUserId);
JsonReply *reply = createAsyncReply("SetupRemoteAccess"); JsonReply *reply = createAsyncReply("SetupRemoteAccess");
m_pairingRequests.insert(cognitoUserId, reply); m_pairingRequests.insert(userId, reply);
connect(reply, &JsonReply::finished, [this, cognitoUserId](){ connect(reply, &JsonReply::finished, [this, userId](){
m_pairingRequests.remove(cognitoUserId); m_pairingRequests.remove(userId);
}); });
return reply; return reply;
} }
@ -375,6 +375,7 @@ void JsonRPCServer::setup()
void JsonRPCServer::processData(const QUuid &clientId, const QByteArray &data) void JsonRPCServer::processData(const QUuid &clientId, const QByteArray &data)
{ {
qCDebug(dcJsonRpcTraffic()) << "Incoming data:" << data; qCDebug(dcJsonRpcTraffic()) << "Incoming data:" << data;
qWarning() << "***F*f+DSF*DA*a+FD*SF*f+ASDF*S";
TransportInterface *interface = qobject_cast<TransportInterface *>(sender()); TransportInterface *interface = qobject_cast<TransportInterface *>(sender());
QJsonParseError error; QJsonParseError error;
@ -447,7 +448,7 @@ void JsonRPCServer::processData(const QUuid &clientId, const QByteArray &data)
handler->setProperty("token", message.value("token").toByteArray()); handler->setProperty("token", message.value("token").toByteArray());
handler->setProperty("transportInterface", reinterpret_cast<qint64>(interface)); handler->setProperty("transportInterface", reinterpret_cast<qint64>(interface));
qCDebug(dcJsonRpc()) << "Got method" << method.toLatin1().data(); qCDebug(dcJsonRpc()) << "Invoking method" << targetNamespace << method.toLatin1().data();
JsonReply *reply; JsonReply *reply;
QMetaObject::invokeMethod(handler, method.toLatin1().data(), Q_RETURN_ARG(JsonReply*, reply), Q_ARG(QVariantMap, params)); QMetaObject::invokeMethod(handler, method.toLatin1().data(), Q_RETURN_ARG(JsonReply*, reply), Q_ARG(QVariantMap, params));
@ -505,15 +506,15 @@ void JsonRPCServer::asyncReplyFinished()
reply->deleteLater(); reply->deleteLater();
} }
void JsonRPCServer::pairingFinished(QString cognitoUserId, int status) void JsonRPCServer::pairingFinished(QString cognitoUserId, int status, const QString &message)
{ {
qWarning() << "pairing finished" << cognitoUserId << status;
JsonReply *reply = m_pairingRequests.take(cognitoUserId); JsonReply *reply = m_pairingRequests.take(cognitoUserId);
if (!reply) { if (!reply) {
return; return;
} }
QVariantMap returns; QVariantMap returns;
returns.insert("status", status); returns.insert("status", status);
returns.insert("message", message);
reply->setData(returns); reply->setData(returns);
reply->finished(); reply->finished();
} }

2
libguh-core/jsonrpc/jsonrpcserver.h
View File

@ -81,7 +81,7 @@ private slots:
void asyncReplyFinished(); void asyncReplyFinished();
void pairingFinished(QString cognitoUserId, int status); void pairingFinished(QString cognitoUserId, int status, const QString &message);
private: private:
QMap<TransportInterface*, bool> m_interfaces; QMap<TransportInterface*, bool> m_interfaces;

3
libguh-core/tcpserver.cpp
View File

@ -114,6 +114,7 @@ void TcpServer::onEncrypted()
void TcpServer::onDataAvailable(QSslSocket * socket, const QByteArray &data) void TcpServer::onDataAvailable(QSslSocket * socket, const QByteArray &data)
{ {
qCDebug(dcTcpServerTraffic()) << "Emitting data available";
QUuid clientId = m_clientList.key(socket); QUuid clientId = m_clientList.key(socket);
emit dataAvailable(clientId, data); emit dataAvailable(clientId, data);
} }
@ -224,11 +225,13 @@ void SslServer::onSocketReadyRead()
m_receiveBuffer.append(socket->readAll()); m_receiveBuffer.append(socket->readAll());
int splitIndex = m_receiveBuffer.indexOf("}\n{"); int splitIndex = m_receiveBuffer.indexOf("}\n{");
while (splitIndex > -1) { while (splitIndex > -1) {
qWarning() << "******" << "emitting dataAcailable" << splitIndex;
emit dataAvailable(socket, m_receiveBuffer.left(splitIndex + 1)); emit dataAvailable(socket, m_receiveBuffer.left(splitIndex + 1));
m_receiveBuffer = m_receiveBuffer.right(m_receiveBuffer.length() - splitIndex - 2); m_receiveBuffer = m_receiveBuffer.right(m_receiveBuffer.length() - splitIndex - 2);
splitIndex = m_receiveBuffer.indexOf("}\n{"); splitIndex = m_receiveBuffer.indexOf("}\n{");
} }
if (m_receiveBuffer.endsWith("}\n")) { if (m_receiveBuffer.endsWith("}\n")) {
qWarning() << "*********" << "emitting dataAvailable" << m_receiveBuffer;
emit dataAvailable(socket, m_receiveBuffer); emit dataAvailable(socket, m_receiveBuffer);
m_receiveBuffer.clear(); m_receiveBuffer.clear();
} }

6
tests/scripts/setupremoteaccess.sh
View File

@ -1,7 +1,7 @@
#!/bin/bash #!/bin/bash
if [ -z $4 ]; then if [ "$3" == "" ]; then
echo "usage: $0 host idToken authToken cognitoId" echo "usage: $0 host userId idToken"
else else
(echo '{"id":1, "method": "JSONRPC.SetupRemoteAccess", "params": { "idToken": "'$2'", "authToken": "'$3'", "cognitoId": "'$4'"}}'; sleep 1) | nc $1 2222 (echo '{"id":1, "method":"JSONRPC.SetupRemoteAccess", "params":{"userId": "'$2'", "idToken":"'$3'"}}'; sleep 1) | nc $1 2222
fi fi