Merge pull request #348 from quotient-im/kitsune-better-basejob

Better BaseJob

1 files changed, 76 insertions, 6 deletions

diff --git a/lib/connectiondata.cpp b/lib/connectiondata.cpp
index 486de03d..a3807fc4 100644
--- a/lib/connectiondata.cpp
+++ b/lib/connectiondata.cpp
@@ -20,27 +20,93 @@
 
 #include "logging.h"
 #include "networkaccessmanager.h"
+#include "jobs/basejob.h"
+
+#include <QtCore/QTimer>
+#include <QtCore/QPointer>
+
+#include <array>
+#include <queue>
 
 using namespace Quotient;
 
-struct ConnectionData::Private {
-    explicit Private(QUrl url) : baseUrl(std::move(url)) {}
+class ConnectionData::Private {
+public:
+    explicit Private(QUrl url) : baseUrl(std::move(url))
+    {
+        rateLimiter.setSingleShot(true);
+    }
 
     QUrl baseUrl;
     QByteArray accessToken;
     QString lastEvent;
+    QString userId;
     QString deviceId;
 
     mutable unsigned int txnCounter = 0;
-    const qint64 id = QDateTime::currentMSecsSinceEpoch();
+    const qint64 txnBase = QDateTime::currentMSecsSinceEpoch();
+
+    QString id() const { return userId + '/' + deviceId; }
+
+    using job_queue_t = std::queue<QPointer<BaseJob>>;
+    std::array<job_queue_t, 2> jobs; // 0 - foreground, 1 - background
+    QTimer rateLimiter;
 };
 
 ConnectionData::ConnectionData(QUrl baseUrl)
     : d(std::make_unique<Private>(std::move(baseUrl)))
-{}
+{
+    // Each lambda invocation below takes no more than one job from the
+    // queues (first foreground, then background) and resumes it; then
+    // restarts the rate limiter timer with duration 0, effectively yielding
+    // to the event loop and then resuming until both queues are empty.
+    QObject::connect(&d->rateLimiter, &QTimer::timeout, [this] {
+        // TODO: Consider moving out all job->sendRequest() invocations to
+        // a dedicated thread
+        d->rateLimiter.setInterval(0);
+        for (auto& q : d->jobs)
+            while (!q.empty()) {
+                auto& job = q.front();
+                q.pop();
+                if (!job || job->error() == BaseJob::Abandoned)
+                    continue;
+                if (job->error() != BaseJob::Pending) {
+                    qCCritical(MAIN)
+                        << "Job" << job
+                        << "is in the wrong status:" << job->status();
+                    Q_ASSERT(false);
+                    job->setStatus(BaseJob::Pending);
+                }
+                job->sendRequest();
+                d->rateLimiter.start();
+                return;
+            }
+        qCDebug(MAIN) << d->id() << "job queues are empty";
+    });
+}
 
 ConnectionData::~ConnectionData() = default;
 
+void ConnectionData::submit(BaseJob* job)
+{
+    Q_ASSERT(job->error() == BaseJob::Pending);
+    if (!d->rateLimiter.isActive()) {
+        job->sendRequest();
+        return;
+    }
+    d->jobs[size_t(job->isBackground())].emplace(job);
+    qCDebug(MAIN) << job << "queued," << d->jobs.front().size() << "+"
+                  << d->jobs.back().size() << "total jobs in" << d->id()
+                  << "queues";
+}
+
+void ConnectionData::limitRate(std::chrono::milliseconds nextCallAfter)
+{
+    qCDebug(MAIN) << "Jobs for" << (d->userId + "/" + d->deviceId)
+                  << "suspended for" << nextCallAfter.count() << "ms";
+    d->rateLimiter.start(nextCallAfter);
+}
+
 QByteArray ConnectionData::accessToken() const { return d->accessToken; }
 
 QUrl ConnectionData::baseUrl() const { return d->baseUrl; }
@@ -75,12 +141,15 @@ void ConnectionData::setPort(int port)
 
 const QString& ConnectionData::deviceId() const { return d->deviceId; }
 
+const QString& ConnectionData::userId() const { return d->userId; }
+
 void ConnectionData::setDeviceId(const QString& deviceId)
 {
     d->deviceId = deviceId;
-    qCDebug(MAIN) << "updated deviceId to" << d->deviceId;
 }
 
+void ConnectionData::setUserId(const QString& userId) { d->userId = userId; }
+
 QString ConnectionData::lastEvent() const { return d->lastEvent; }
 
 void ConnectionData::setLastEvent(QString identifier)
@@ -90,5 +159,6 @@ void ConnectionData::setLastEvent(QString identifier)
 
 QByteArray ConnectionData::generateTxnId() const
 {
-    return QByteArray::number(d->id) + 'q' + QByteArray::number(++d->txnCounter);
+    return d->deviceId.toLatin1() + QByteArray::number(d->txnBase)
+           + QByteArray::number(++d->txnCounter);
 }