QXmppStream: Report packet sending result using QFutures

7 files changed, 207 insertions, 24 deletions

diff --git a/src/base/QXmppGlobal.h.in b/src/base/QXmppGlobal.h.in
index 5f94196d..ac249558 100644
--- a/src/base/QXmppGlobal.h.in
+++ b/src/base/QXmppGlobal.h.in
@@ -23,7 +23,6 @@
  *
  */
 
-
 #ifndef QXMPPGLOBAL_H
 #define QXMPPGLOBAL_H
 
@@ -77,5 +76,22 @@ inline QLatin1String QXmppVersion()
 #define QT_WARNING_DISABLE_DEPRECATED
 #endif
 
-#endif // QXMPPGLOBAL_H
+///
+/// \namespace QXmpp
+///
+/// Contains global functions and enumerations.
+///
+namespace QXmpp {
+
+///
+/// The state of an outgoing packet.
+///
+enum PacketState : quint8 {
+    Sent,          ///< The packet has been written to the socket.
+    Acknowledged,  ///< The packet has been acknowledged by the other peer using Stream Management.
+    NotSent,       ///< The packet could not be sent (e.g. connection broke or user disconnected).
+};
+
+}
 
+#endif  // QXMPPGLOBAL_H
diff --git a/src/base/QXmppPacket.cpp b/src/base/QXmppPacket.cpp
new file mode 100644
index 00000000..86b28ed2
--- /dev/null
+++ b/src/base/QXmppPacket.cpp
@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2008-2021 The QXmpp developers
+ *
+ * Authors:
+ *  Linus Jahn
+ *
+ * Source:
+ *  https://github.com/qxmpp-project/qxmpp
+ *
+ * This file is a part of QXmpp library.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ */
+
+#include "QXmppPacket_p.h"
+#include "QXmppStanza.h"
+
+#include <QFuture>
+#include <QXmlStreamWriter>
+
+/// \cond
+QXmppPacket::QXmppPacket(const QXmppStanza &stanza)
+    : m_interface(new QFutureInterface<QXmpp::PacketState>(QFutureInterfaceBase::Started)),
+      m_isXmppStanza(stanza.isXmppStanza())
+{
+    QXmlStreamWriter xmlStream(&m_data);
+    stanza.toXml(&xmlStream);
+}
+
+QByteArray QXmppPacket::data() const
+{
+    return m_data;
+}
+
+bool QXmppPacket::isXmppStanza() const
+{
+    return m_isXmppStanza;
+}
+
+QFuture<QXmpp::PacketState> QXmppPacket::future()
+{
+    return m_interface->future();
+}
+
+void QXmppPacket::reportFinished()
+{
+    m_interface->reportFinished();
+}
+
+void QXmppPacket::reportResult(QXmpp::PacketState result)
+{
+    m_interface->reportResult(result);
+}
+/// \endcond
diff --git a/src/base/QXmppPacket_p.h b/src/base/QXmppPacket_p.h
new file mode 100644
index 00000000..1ec2f592
--- /dev/null
+++ b/src/base/QXmppPacket_p.h
@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2008-2021 The QXmpp developers
+ *
+ * Authors:
+ *  Linus Jahn
+ *
+ * Source:
+ *  https://github.com/qxmpp-project/qxmpp
+ *
+ * This file is a part of QXmpp library.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ */
+
+#ifndef QXMPPPACKET_H
+#define QXMPPPACKET_H
+
+#include "QXmppGlobal.h"
+
+#include <QFutureInterface>
+#include <QSharedPointer>
+
+class QXmppStanza;
+
+class QXmppPacket
+{
+public:
+    QXmppPacket(const QXmppStanza &stanza);
+
+    QByteArray data() const;
+    bool isXmppStanza() const;
+
+    QFuture<QXmpp::PacketState> future();
+
+    void reportFinished();
+    void reportResult(QXmpp::PacketState);
+
+private:
+    QSharedPointer<QFutureInterface<QXmpp::PacketState>> m_interface;
+    QByteArray m_data;
+    bool m_isXmppStanza;
+};
+
+#endif  // QXMPPPACKET_H
diff --git a/src/base/QXmppStream.cpp b/src/base/QXmppStream.cpp
index bffaf4c7..7426f8b0 100644
--- a/src/base/QXmppStream.cpp
+++ b/src/base/QXmppStream.cpp
@@ -26,12 +26,15 @@
 
 #include "QXmppConstants_p.h"
 #include "QXmppLogger.h"
+#include "QXmppPacket_p.h"
 #include "QXmppStanza.h"
 #include "QXmppStreamManagement_p.h"
 #include "QXmppUtils.h"
 
 #include <QBuffer>
 #include <QDomDocument>
+#include <QFuture>
+#include <QFutureInterface>
 #include <QHostAddress>
 #include <QMap>
 #include <QRegularExpression>
@@ -147,22 +150,28 @@ bool QXmppStream::sendData(const QByteArray &data)
 ///
 /// Sends an XMPP packet to the peer.
 ///
-/// \param packet
+/// \param stanza
 ///
-bool QXmppStream::sendPacket(const QXmppStanza &packet)
+bool QXmppStream::sendPacket(const QXmppStanza &stanza)
 {
-    // prepare packet
-    QByteArray data;
-    QXmlStreamWriter xmlStream(&data);
-    packet.toXml(&xmlStream);
+    // the first result is always reported immediately
+    return send(stanza).resultAt(0) != QXmpp::NotSent;
+}
 
-    // send packet
-    bool success = sendData(data);
+///
+/// Sends an XMPP packet to the peer.
+///
+/// \since QXmpp 1.5
+///
+QFuture<QXmpp::PacketState> QXmppStream::send(const QXmppStanza &stanza)
+{
+    QXmppPacket packet(stanza);
+    sendPacket(packet);
 
     // handle stream management
-    d->streamManager.handlePacketSent(packet, data);
+    d->streamManager.handlePacketSent(packet);
 
-    return success;
+    return packet.future();
 }
 
 ///
@@ -334,6 +343,15 @@ void QXmppStream::processData(const QString &data)
     }
 }
 
+void QXmppStream::sendPacket(QXmppPacket &packet)
+{
+    if (sendData(packet.data())) {
+        packet.reportResult(QXmpp::Sent);
+    } else {
+        packet.reportResult(QXmpp::NotSent);
+    }
+}
+
 ///
 /// Enables Stream Management acks / reqs (\xep{0198}).
 ///
diff --git a/src/base/QXmppStream.h b/src/base/QXmppStream.h
index 2530dea8..65547814 100644
--- a/src/base/QXmppStream.h
+++ b/src/base/QXmppStream.h
@@ -31,7 +31,10 @@
 #include <QObject>
 
 class QDomElement;
+template<typename T>
+class QFuture;
 class QSslSocket;
+class QXmppPacket;
 class QXmppStanza;
 class QXmppStreamPrivate;
 
@@ -47,7 +50,9 @@ public:
     ~QXmppStream() override;
 
     virtual bool isConnected() const;
+
     bool sendPacket(const QXmppStanza &);
+    QFuture<QXmpp::PacketState> send(const QXmppStanza &);
 
     void resetPacketCache();
 
@@ -92,10 +97,12 @@ private Q_SLOTS:
     void _q_socketReadyRead();
 
 private:
-    void processData(const QString &data);
-
+    friend class QXmppStreamManager;
     friend class tst_QXmppStream;
 
+    void processData(const QString &data);
+    void sendPacket(QXmppPacket &packet);
+
     QXmppStreamPrivate *const d;
 };
 
diff --git a/src/base/QXmppStreamManagement.cpp b/src/base/QXmppStreamManagement.cpp
index 9b59cd8a..4d5f3540 100644
--- a/src/base/QXmppStreamManagement.cpp
+++ b/src/base/QXmppStreamManagement.cpp
@@ -23,6 +23,7 @@
 
 #include "QXmppConstants_p.h"
 #include "QXmppGlobal.h"
+#include "QXmppPacket_p.h"
 #include "QXmppStanza_p.h"
 #include "QXmppStream.h"
 #include "QXmppStreamManagement_p.h"
@@ -332,6 +333,16 @@ QXmppStreamManager::QXmppStreamManager(QXmppStream *stream)
 {
 }
 
+QXmppStreamManager::~QXmppStreamManager()
+{
+    resetCache();
+}
+
+bool QXmppStreamManager::enabled() const
+{
+    return m_enabled;
+}
+
 unsigned int QXmppStreamManager::lastIncomingSequenceNumber() const
 {
     return m_lastIncomingSequenceNumber;
@@ -340,6 +351,9 @@ unsigned int QXmppStreamManager::lastIncomingSequenceNumber() const
 void QXmppStreamManager::handleDisconnect()
 {
     m_enabled = false;
+    for (auto &packet : m_unacknowledgedStanzas) {
+        packet.reportResult(QXmpp::NotSent);
+    }
 }
 
 void QXmppStreamManager::handleStart()
@@ -347,11 +361,13 @@ void QXmppStreamManager::handleStart()
     m_enabled = false;
 }
 
-void QXmppStreamManager::handlePacketSent(const QXmppStanza &packet, const QByteArray &data)
+void QXmppStreamManager::handlePacketSent(QXmppPacket &packet)
 {
     if (m_enabled && packet.isXmppStanza()) {
-        m_unacknowledgedStanzas.insert(++m_lastOutgoingSequenceNumber, data);
+        m_unacknowledgedStanzas.insert(++m_lastOutgoingSequenceNumber, packet);
         sendAcknowledgementRequest();
+    } else {
+        packet.reportFinished();
     }
 }
 
@@ -384,12 +400,12 @@ void QXmppStreamManager::enableStreamManagement(bool resetSequenceNumber)
 
         // resend unacked stanzas
         if (!m_unacknowledgedStanzas.isEmpty()) {
-            const auto oldUnackedStanzas = m_unacknowledgedStanzas;
+            auto oldUnackedStanzas = m_unacknowledgedStanzas;
             m_unacknowledgedStanzas.clear();
 
-            for (const auto &value : oldUnackedStanzas) {
-                m_unacknowledgedStanzas.insert(++m_lastOutgoingSequenceNumber, value);
-                stream->sendData(value);
+            for (auto &packet : oldUnackedStanzas) {
+                m_unacknowledgedStanzas.insert(++m_lastOutgoingSequenceNumber, packet);
+                stream->sendPacket(packet);
             }
 
             sendAcknowledgementRequest();
@@ -397,8 +413,8 @@ void QXmppStreamManager::enableStreamManagement(bool resetSequenceNumber)
     } else {
         // resend unacked stanzas
         if (!m_unacknowledgedStanzas.isEmpty()) {
-            for (const auto &value : std::as_const(m_unacknowledgedStanzas)) {
-                stream->sendData(value);
+            for (auto &packet : m_unacknowledgedStanzas) {
+                stream->sendPacket(packet);
             }
 
             sendAcknowledgementRequest();
@@ -410,6 +426,8 @@ void QXmppStreamManager::setAcknowledgedSequenceNumber(unsigned int sequenceNumb
 {
     for (auto it = m_unacknowledgedStanzas.begin(); it != m_unacknowledgedStanzas.end();) {
         if (it.key() <= sequenceNumber) {
+            it->reportResult(QXmpp::Acknowledged);
+            it->reportFinished();
             it = m_unacknowledgedStanzas.erase(it);
         } else {
             break;
@@ -458,6 +476,11 @@ void QXmppStreamManager::sendAcknowledgementRequest()
 
 void QXmppStreamManager::resetCache()
 {
+    for (auto &packet : m_unacknowledgedStanzas) {
+        packet.reportResult(QXmpp::NotSent);
+        packet.reportFinished();
+    }
+
     m_unacknowledgedStanzas.clear();
 }
 /// \endcond
diff --git a/src/base/QXmppStreamManagement_p.h b/src/base/QXmppStreamManagement_p.h
index fb79c7b0..1a528c0c 100644
--- a/src/base/QXmppStreamManagement_p.h
+++ b/src/base/QXmppStreamManagement_p.h
@@ -31,6 +31,7 @@
 #include <QXmlStreamWriter>
 
 class QXmppStream;
+class QXmppPacket;
 
 //
 //  W A R N I N G
@@ -189,12 +190,14 @@ class QXmppStreamManager
 {
 public:
     explicit QXmppStreamManager(QXmppStream *stream);
+    ~QXmppStreamManager();
 
+    bool enabled() const;
     unsigned int lastIncomingSequenceNumber() const;
 
     void handleDisconnect();
     void handleStart();
-    void handlePacketSent(const QXmppStanza &packet, const QByteArray &data);
+    void handlePacketSent(QXmppPacket &packet);
     bool handleStanza(const QDomElement &stanza);
 
     void resetCache();
@@ -210,7 +213,7 @@ private:
     QXmppStream *stream;
 
     bool m_enabled = false;
-    QMap<unsigned int, QByteArray> m_unacknowledgedStanzas;
+    QMap<unsigned int, QXmppPacket> m_unacknowledgedStanzas;
     unsigned int m_lastOutgoingSequenceNumber = 0;
     unsigned int m_lastIncomingSequenceNumber = 0;
 };