Add reporting of IQ responses with QFutures

5 files changed, 179 insertions, 2 deletions

diff --git a/src/base/QXmppStream.cpp b/src/base/QXmppStream.cpp
index 7426f8b0..56f3afda 100644
--- a/src/base/QXmppStream.cpp
+++ b/src/base/QXmppStream.cpp
@@ -25,6 +25,7 @@
 #include "QXmppStream.h"
 
 #include "QXmppConstants_p.h"
+#include "QXmppIq.h"
 #include "QXmppLogger.h"
 #include "QXmppPacket_p.h"
 #include "QXmppStanza.h"
@@ -35,6 +36,7 @@
 #include <QDomDocument>
 #include <QFuture>
 #include <QFutureInterface>
+#include <QFutureWatcher>
 #include <QHostAddress>
 #include <QMap>
 #include <QRegularExpression>
@@ -47,6 +49,45 @@
 static bool randomSeeded = false;
 #endif
 
+class IqState : public QFutureInterface<QXmppStream::IqResult>
+{
+    Q_DISABLE_COPY(IqState)
+
+public:
+    IqState(QFuture<QXmpp::PacketState> sendFuture, bool streamManagementUsed)
+        : QFutureInterface<QXmppStream::IqResult>(QFutureInterfaceBase::Started),
+          m_sendFuture(std::move(sendFuture)),
+          m_streamManagementUsed(streamManagementUsed)
+    {
+        auto *watcher = new QFutureWatcher<QXmpp::PacketState>();
+        QObject::connect(watcher, &QFutureWatcher<QXmpp::PacketState>::finished, [=]() {
+            const auto result = watcher->future().results().last();
+
+            switch (result) {
+            case QXmpp::Acknowledged:
+            case QXmpp::Sent:
+                break;
+            case QXmpp::NotSent:
+                reportResult(result);
+                reportFinished();
+                break;
+            }
+
+            watcher->deleteLater();
+        });
+        watcher->setFuture(m_sendFuture);
+    }
+
+    bool isStreamManagementUsed() const
+    {
+        return m_streamManagementUsed;
+    }
+
+private:
+    QFuture<QXmpp::PacketState> m_sendFuture;
+    bool m_streamManagementUsed;
+};
+
 class QXmppStreamPrivate
 {
 public:
@@ -60,6 +101,9 @@ public:
 
     // stream management
     QXmppStreamManager streamManager;
+
+    // iq response handling
+    QMap<QString, IqState *> runningIqs;
 };
 
 QXmppStreamPrivate::QXmppStreamPrivate(QXmppStream *stream)
@@ -69,6 +113,17 @@ QXmppStreamPrivate::QXmppStreamPrivate(QXmppStream *stream)
 }
 
 ///
+/// \typedef QXmppStream::IqResult
+///
+/// Contains a QDomElement containing the IQ response or if the request couldn't
+/// be sent a QXmpp::PacketState.
+///
+/// \warning THIS API IS NOT FINALIZED YET!
+///
+/// \since QXmpp 1.5
+///
+
+///
 /// Constructs a base XMPP stream.
 ///
 /// \param parent
@@ -91,6 +146,7 @@ QXmppStream::QXmppStream(QObject *parent)
 ///
 QXmppStream::~QXmppStream()
 {
+    cancelOngoingIqs();
     delete d;
 }
 
@@ -175,6 +231,54 @@ QFuture<QXmpp::PacketState> QXmppStream::send(const QXmppStanza &stanza)
 }
 
 ///
+/// Sends an IQ packet and returns the response asynchronously.
+///
+/// \warning THIS API IS NOT FINALIZED YET!
+///
+/// \since QXmpp 1.5
+///
+QFuture<QXmppStream::IqResult> QXmppStream::sendIq(const QXmppIq &iq)
+{
+    if (iq.id().isEmpty()) {
+        warning(QStringLiteral("QXmppStream::sendIq() error: ID is empty. Using random ID."));
+        auto newIq = iq;
+        newIq.setId(QXmppUtils::generateStanzaUuid());
+        return sendIq(newIq);
+    }
+    if (d->runningIqs.contains(iq.id())) {
+        warning(QStringLiteral("QXmppStream::sendIq() error:"
+                               "The IQ's ID (\"%1\") is already in use. Using random ID.")
+                    .arg(iq.id()));
+        auto newIq = iq;
+        newIq.setId(QXmppUtils::generateStanzaUuid());
+        return sendIq(newIq);
+    }
+
+    auto *interface = new IqState(send(iq), d->streamManager.enabled());
+
+    if (!interface->isFinished()) {
+        d->runningIqs.insert(iq.id(), interface);
+    }
+
+    return interface->future();
+}
+
+///
+/// Cancels all ongoing IQ requests and reports QXmpp::NotSent.
+///
+/// \since QXmpp 1.5
+///
+void QXmppStream::cancelOngoingIqs()
+{
+    for (auto *state : std::as_const(d->runningIqs)) {
+        state->reportResult(QXmpp::NotSent);
+        state->reportFinished();
+        delete state;
+    }
+    d->runningIqs.clear();
+}
+
+///
 /// Resets the stream management packages cache.
 ///
 /// This can be done to prevent that packages from the last connection are being
@@ -330,7 +434,7 @@ void QXmppStream::processData(const QString &data)
     auto stanza = doc.documentElement().firstChildElement();
     for (; !stanza.isNull(); stanza = stanza.nextSiblingElement()) {
         // handle possible stream management packets first
-        if (d->streamManager.handleStanza(stanza))
+        if (d->streamManager.handleStanza(stanza) || handleIqResponse(stanza))
             continue;
 
         // process all other kinds of packets
@@ -352,6 +456,33 @@ void QXmppStream::sendPacket(QXmppPacket &packet)
     }
 }
 
+bool QXmppStream::handleIqResponse(const QDomElement &stanza)
+{
+    if (stanza.tagName() != QStringLiteral("iq")) {
+        return false;
+    }
+
+    // only accept "result" and "error" types
+    const auto iqType = stanza.attribute(QStringLiteral("type"));
+    if (iqType != QStringLiteral("result") && iqType != QStringLiteral("error")) {
+        return false;
+    }
+
+    if (auto itr = d->runningIqs.find(stanza.attribute(QStringLiteral("id")));
+        itr != d->runningIqs.end()) {
+
+        auto *state = itr.value();
+        state->reportResult(stanza);
+        state->reportFinished();
+        delete state;
+
+        d->runningIqs.erase(itr);
+        return true;
+    }
+
+    return false;
+}
+
 ///
 /// Enables Stream Management acks / reqs (\xep{0198}).
 ///
diff --git a/src/base/QXmppStream.h b/src/base/QXmppStream.h
index 65547814..91bf7d09 100644
--- a/src/base/QXmppStream.h
+++ b/src/base/QXmppStream.h
@@ -27,6 +27,8 @@
 
 #include "QXmppLogger.h"
 
+#include <variant>
+
 #include <QAbstractSocket>
 #include <QObject>
 
@@ -34,6 +36,7 @@ class QDomElement;
 template<typename T>
 class QFuture;
 class QSslSocket;
+class QXmppIq;
 class QXmppPacket;
 class QXmppStanza;
 class QXmppStreamPrivate;
@@ -54,6 +57,10 @@ public:
     bool sendPacket(const QXmppStanza &);
     QFuture<QXmpp::PacketState> send(const QXmppStanza &);
 
+    using IqResult = std::variant<QDomElement, QXmpp::PacketState>;
+    QFuture<IqResult> sendIq(const QXmppIq &);
+    void cancelOngoingIqs();
+
     void resetPacketCache();
 
 Q_SIGNALS:
@@ -102,6 +109,7 @@ private:
 
     void processData(const QString &data);
     void sendPacket(QXmppPacket &packet);
+    bool handleIqResponse(const QDomElement &);
 
     QXmppStreamPrivate *const d;
 };
diff --git a/src/client/QXmppClient.cpp b/src/client/QXmppClient.cpp
index 6933a0a6..4f8bfc26 100644
--- a/src/client/QXmppClient.cpp
+++ b/src/client/QXmppClient.cpp
@@ -38,6 +38,7 @@
 #include "QXmppVCardManager.h"
 #include "QXmppVersionManager.h"
 
+#include <QDomElement>
 #include <QFuture>
 #include <QSslSocket>
 #include <QTimer>
@@ -109,6 +110,14 @@ QStringList QXmppClientPrivate::discoveryFeatures()
 }
 /// \endcond
 
+///
+/// \typedef QXmppClient::IqResult
+///
+/// Result of an IQ request, either contains the QDomElement of the IQ answer
+/// (with type 'error' or 'result') or it contains the packet error, if the
+/// request couldn't be sent.
+///
+
 /// Creates a QXmppClient object.
 /// \param parent is passed to the QObject's constructor.
 /// The default value is 0.
@@ -333,6 +342,18 @@ QFuture<QXmpp::PacketState> QXmppClient::send(const QXmppStanza &stanza)
     return d->stream->send(stanza);
 }
 
+///
+/// Sends an IQ packet and returns the response asynchronously.
+///
+/// \warning THIS API IS NOT FINALIZED YET!
+///
+/// \since QXmpp 1.5
+///
+QFuture<QXmppClient::IqResult> QXmppClient::sendIq(const QXmppIq &iq)
+{
+    return d->stream->sendIq(iq);
+}
+
 /// Disconnects the client and the current presence of client changes to
 /// QXmppPresence::Unavailable and status text changes to "Logged out".
 ///
diff --git a/src/client/QXmppClient.h b/src/client/QXmppClient.h
index 56da95ee..202c9c5a 100644
--- a/src/client/QXmppClient.h
+++ b/src/client/QXmppClient.h
@@ -221,6 +221,9 @@ public:
 
     QFuture<QXmpp::PacketState> send(const QXmppStanza &);
 
+    using IqResult = std::variant<QDomElement, QXmpp::PacketState>;
+    QFuture<IqResult> sendIq(const QXmppIq &);
+
 #if QXMPP_DEPRECATED_SINCE(1, 1)
     QT_DEPRECATED_X("Use QXmppClient::findExtension<QXmppRosterManager>() instead")
     QXmppRosterManager &rosterManager();
diff --git a/src/client/QXmppOutgoingClient.cpp b/src/client/QXmppOutgoingClient.cpp
index a8bfb62d..6479f334 100644
--- a/src/client/QXmppOutgoingClient.cpp
+++ b/src/client/QXmppOutgoingClient.cpp
@@ -213,6 +213,20 @@ QXmppOutgoingClient::QXmppOutgoingClient(QObject *parent)
 
     connect(this, &QXmppStream::connected, this, &QXmppOutgoingClient::pingStart);
     connect(this, &QXmppStream::disconnected, this, &QXmppOutgoingClient::pingStop);
+
+    // IQ response handling
+    connect(this, &QXmppStream::connected, this, [=]() {
+        if (!d->streamResumed) {
+            // we can't expect a response because this is a new stream
+            cancelOngoingIqs();
+        }
+    });
+    connect(this, &QXmppStream::disconnected, this, [=]() {
+        if (!d->canResume) {
+            // this stream can't be resumed; we can cancel all ongoing IQs
+            cancelOngoingIqs();
+        }
+    });
 }
 
 /// Destroys an outgoing client stream.
@@ -510,7 +524,7 @@ void QXmppOutgoingClient::handleStanza(const QDomElement &nodeRecv)
         d->bindModeAvailable = (features.bindMode() != QXmppStreamFeatures::Disabled);
         d->streamManagementAvailable = (features.streamManagementMode() != QXmppStreamFeatures::Disabled);
 
-        // chech whether the stream can be resumed
+        // check whether the stream can be resumed
         if (d->streamManagementAvailable && d->canResume) {
             d->isResuming = true;
             QXmppStreamManagementResume streamManagementResume(lastIncomingSequenceNumber(), d->smId);