From b4df9eb5f1f4a0ac2b1272fc34d4b8aad473008b Mon Sep 17 00:00:00 2001 From: Christian Mollekopf Date: Tue, 5 Jul 2016 15:22:10 +0200 Subject: Prepare for making the resource status available --- common/changereplay.cpp | 1 + common/changereplay.h | 1 + common/commands/notification.fbs | 7 ++-- common/domain/applicationdomaintype.h | 62 +++++++++++++++++++++++++++++++---- common/genericresource.cpp | 40 +++++++++++++++++++--- common/listener.cpp | 4 +-- common/notification.h | 13 ++++++++ common/resourceaccess.cpp | 49 +++++++++++++++++---------- common/resourceaccess.h | 10 +++++- tests/dummyresourcebenchmark.cpp | 2 +- 10 files changed, 151 insertions(+), 38 deletions(-) diff --git a/common/changereplay.cpp b/common/changereplay.cpp index 0096bd0..78c0ff5 100644 --- a/common/changereplay.cpp +++ b/common/changereplay.cpp @@ -86,6 +86,7 @@ KAsync::Job ChangeReplay::replayNextRevision() if (lastReplayedRevision < topRevision) { Trace() << "Changereplay from " << lastReplayedRevision << " to " << topRevision; + emit replayingChanges(); qint64 revision = lastReplayedRevision + 1; const auto uid = Storage::getUidFromRevision(mainStoreTransaction, revision); const auto type = Storage::getTypeFromRevision(mainStoreTransaction, revision); diff --git a/common/changereplay.h b/common/changereplay.h index aba3dd0..6c1c1db 100644 --- a/common/changereplay.h +++ b/common/changereplay.h @@ -45,6 +45,7 @@ public: signals: void changesReplayed(); + void replayingChanges(); public slots: void revisionChanged(); diff --git a/common/commands/notification.fbs b/common/commands/notification.fbs index 8750ff5..c82fad3 100644 --- a/common/commands/notification.fbs +++ b/common/commands/notification.fbs @@ -1,13 +1,10 @@ namespace Sink.Commands; -enum NotificationType : byte { Shutdown = 1, Status, Warning, Progress, Inspection, RevisionUpdate } -enum NotificationCode : byte { Success = 0, Failure = 1, UserCode } - table Notification { - type: NotificationType = Status; + type: int = 0; //See notification.h identifier: string; //An identifier that links back to the something related to the notification (e.g. an entity id or a command id) message: string; - code: int = 0; //Of type NotificationCode + code: int = 0; //See notification.h } root_type Notification; diff --git a/common/domain/applicationdomaintype.h b/common/domain/applicationdomaintype.h index 849c3e2..5efb936 100644 --- a/common/domain/applicationdomaintype.h +++ b/common/domain/applicationdomaintype.h @@ -52,6 +52,13 @@ void setExtracted##NAME(const TYPE &value) { setProperty(NAME::name, QVariant::fromValue(value)); } \ TYPE get##NAME() const { return getProperty(NAME::name).value(); } \ +#define SINK_STATUS_PROPERTY(TYPE, NAME, LOWERCASENAME) \ + struct NAME { \ + static constexpr const char *name = #LOWERCASENAME; \ + typedef TYPE Type; \ + }; \ + void setStatus##NAME(const TYPE &value) { setProperty(NAME::name, QVariant::fromValue(value)); } \ + TYPE get##NAME() const { return getProperty(NAME::name).value(); } \ #define SINK_BLOB_PROPERTY(NAME, LOWERCASENAME) \ struct NAME { \ @@ -76,6 +83,14 @@ namespace Sink { namespace ApplicationDomain { +struct SINK_EXPORT Error { + +}; + +struct SINK_EXPORT Progress { + +}; + /** * The domain type interface has two purposes: * * provide a unified interface to read buffers (for zero-copy reading) @@ -214,6 +229,38 @@ struct SINK_EXPORT Mail : public Entity { SINK_PROPERTY(bool, Sent, sent); }; +/** + * The status of an account or resource. + * + * It is set as follows: + * * By default the status is offline. + * * If a connection to the server could be established the status is Connected. + * * If an error occurred that keeps the resource from operating (so non transient), the resource enters the error state. + * * If a long running operation is started the resource goes to the busy state (and return to the previous state after that). + */ +enum SINK_EXPORT Status { + OfflineStatus, + ConnectedStatus, + BusyStatus, + ErrorStatus +}; + +struct SINK_EXPORT SinkAccount : public ApplicationDomainType { + typedef QSharedPointer Ptr; + explicit SinkAccount(const QByteArray &resourceInstanceIdentifier, const QByteArray &identifier, qint64 revision, const QSharedPointer &adaptor); + explicit SinkAccount(const QByteArray &identifier); + SinkAccount(); + virtual ~SinkAccount(); + + SINK_PROPERTY(QString, Name, name); + SINK_PROPERTY(QString, Icon, icon); + SINK_PROPERTY(QString, AccountType, accountType); + SINK_STATUS_PROPERTY(int, Status, status); + SINK_STATUS_PROPERTY(ApplicationDomain::Error, Error, error); + SINK_STATUS_PROPERTY(ApplicationDomain::Progress, Progress, progress); +}; + + /** * Represents an sink resource. * @@ -226,14 +273,13 @@ struct SINK_EXPORT SinkResource : public ApplicationDomainType { explicit SinkResource(const QByteArray &identifier); SinkResource(); virtual ~SinkResource(); -}; -struct SINK_EXPORT SinkAccount : public ApplicationDomainType { - typedef QSharedPointer Ptr; - explicit SinkAccount(const QByteArray &resourceInstanceIdentifier, const QByteArray &identifier, qint64 revision, const QSharedPointer &adaptor); - explicit SinkAccount(const QByteArray &identifier); - SinkAccount(); - virtual ~SinkAccount(); + SINK_REFERENCE_PROPERTY(SinkAccount, Account, account); + SINK_PROPERTY(QString, ResourceType, resourceType); + SINK_PROPERTY(QByteArrayList, Capabilities, capabilities); + SINK_STATUS_PROPERTY(int, Status, status); + SINK_STATUS_PROPERTY(ApplicationDomain::Error, Error, error); + SINK_STATUS_PROPERTY(ApplicationDomain::Progress, Progress, progress); }; struct SINK_EXPORT Identity : public ApplicationDomainType { @@ -330,3 +376,5 @@ Q_DECLARE_METATYPE(Sink::ApplicationDomain::SinkAccount) Q_DECLARE_METATYPE(Sink::ApplicationDomain::SinkAccount::Ptr) Q_DECLARE_METATYPE(Sink::ApplicationDomain::Identity) Q_DECLARE_METATYPE(Sink::ApplicationDomain::Identity::Ptr) +Q_DECLARE_METATYPE(Sink::ApplicationDomain::Error) +Q_DECLARE_METATYPE(Sink::ApplicationDomain::Progress) diff --git a/common/genericresource.cpp b/common/genericresource.cpp index c06c22a..5522174 100644 --- a/common/genericresource.cpp +++ b/common/genericresource.cpp @@ -260,18 +260,18 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra [=]() { Log_area("resource.inspection") << "Inspection was successful: " << inspectionType << inspectionId << entityId; Sink::Notification n; - n.type = Sink::Commands::NotificationType_Inspection; + n.type = Sink::Notification::Inspection; n.id = inspectionId; - n.code = Sink::Commands::NotificationCode_Success; + n.code = Sink::Notification::Success; emit notify(n); }, [=](int code, const QString &message) { Warning_area("resource.inspection") << "Inspection failed: " << inspectionType << inspectionId << entityId << message; Sink::Notification n; - n.type = Sink::Commands::NotificationType_Inspection; + n.type = Sink::Notification::Inspection; n.message = message; n.id = inspectionId; - n.code = Sink::Commands::NotificationCode_Failure; + n.code = Sink::Notification::Failure; emit notify(n); }) .exec(); @@ -283,6 +283,23 @@ GenericResource::GenericResource(const QByteArray &resourceType, const QByteArra QObject::connect(mPipeline.data(), &Pipeline::revisionUpdated, this, &Resource::revisionUpdated); mClientLowerBoundRevision = mPipeline->cleanedUpRevision(); + QObject::connect(mChangeReplay.data(), &ChangeReplay::replayingChanges, [this]() { + Sink::Notification n; + n.id = "changereplay"; + n.type = Sink::Notification::Status; + n.message = "Replaying changes."; + n.code = Sink::ApplicationDomain::BusyStatus; + emit notify(n); + }); + QObject::connect(mChangeReplay.data(), &ChangeReplay::changesReplayed, [this]() { + Sink::Notification n; + n.id = "changereplay"; + n.type = Sink::Notification::Status; + n.message = "All changes have been replayed."; + n.code = Sink::ApplicationDomain::ConnectedStatus; + emit notify(n); + }); + mCommitQueueTimer.setInterval(sCommitInterval); mCommitQueueTimer.setSingleShot(true); QObject::connect(&mCommitQueueTimer, &QTimer::timeout, &mUserQueue, &MessageQueue::commit); @@ -399,12 +416,27 @@ void GenericResource::processCommand(int commandId, const QByteArray &data) KAsync::Job GenericResource::synchronizeWithSource() { return KAsync::start([this](KAsync::Future &future) { + + Sink::Notification n; + n.id = "sync"; + n.type = Sink::Notification::Status; + n.message = "Synchronization has started."; + n.code = Sink::ApplicationDomain::BusyStatus; + emit notify(n); + Log() << " Synchronizing"; // Changereplay would deadlock otherwise when trying to open the synchronization store enableChangeReplay(false); mSynchronizer->synchronize() .then([this, &future]() { Log() << "Done Synchronizing"; + Sink::Notification n; + n.id = "sync"; + n.type = Sink::Notification::Status; + n.message = "Synchronization has ended."; + n.code = Sink::ApplicationDomain::ConnectedStatus; + emit notify(n); + enableChangeReplay(true); future.setFinished(); }, [this, &future](int errorCode, const QString &error) { diff --git a/common/listener.cpp b/common/listener.cpp index 84afe16..d2fc510 100644 --- a/common/listener.cpp +++ b/common/listener.cpp @@ -330,7 +330,7 @@ qint64 Listener::lowerBoundRevision() void Listener::quit() { // Broadcast shutdown notifications to open clients, so they don't try to restart the resource - auto command = Sink::Commands::CreateNotification(m_fbb, Sink::Commands::NotificationType::NotificationType_Shutdown); + auto command = Sink::Commands::CreateNotification(m_fbb, Sink::Notification::Shutdown); Sink::Commands::FinishNotificationBuffer(m_fbb, command); for (Client &client : m_connections) { if (client.socket && client.socket->isOpen()) { @@ -418,7 +418,7 @@ void Listener::notify(const Sink::Notification ¬ification) auto messageString = m_fbb.CreateString(notification.message.toUtf8().constData(), notification.message.toUtf8().size()); auto idString = m_fbb.CreateString(notification.id.constData(), notification.id.size()); Sink::Commands::NotificationBuilder builder(m_fbb); - builder.add_type(static_cast(notification.type)); + builder.add_type(notification.type); builder.add_code(notification.code); builder.add_identifier(idString); builder.add_message(messageString); diff --git a/common/notification.h b/common/notification.h index 0eb796d..0a267e6 100644 --- a/common/notification.h +++ b/common/notification.h @@ -30,6 +30,19 @@ namespace Sink { class SINK_EXPORT Notification { public: + enum NoticationType { + Shutdown, + Status, + Warning, + Progress, + Inspection, + RevisionUpdate + }; + enum InspectionCode { + Success, + Failure + }; + QByteArray id; int type; QString message; diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp index d3bd85f..95b4a7e 100644 --- a/common/resourceaccess.cpp +++ b/common/resourceaccess.cpp @@ -230,6 +230,7 @@ KAsync::Job ResourceAccess::Private::initializeSocket() ResourceAccess::ResourceAccess(const QByteArray &resourceInstanceIdentifier, const QByteArray &resourceType) : ResourceAccessInterface(), d(new Private(resourceType, resourceInstanceIdentifier, this)) { + mResourceStatus = Sink::ApplicationDomain::OfflineStatus; Trace() << "Starting access"; } @@ -513,6 +514,22 @@ void ResourceAccess::readResourceMessage() } } +static Sink::Notification getNotification(const Sink::Commands::Notification *buffer) +{ + Sink::Notification n; + if (buffer->identifier()) { + // Don't use fromRawData, the buffer is gone once we invoke emit notification + n.id = BufferUtils::extractBufferCopy(buffer->identifier()); + } + if (buffer->message()) { + // Don't use fromRawData, the buffer is gone once we invoke emit notification + n.message = BufferUtils::extractBufferCopy(buffer->message()); + } + n.type = buffer->type(); + n.code = buffer->code(); + return n; +} + bool ResourceAccess::processMessageBuffer() { static const int headerSize = Commands::headerSize(); @@ -535,7 +552,7 @@ bool ResourceAccess::processMessageBuffer() auto buffer = Commands::GetRevisionUpdate(d->partialMessageBuffer.constData() + headerSize); Trace() << QString("Revision updated to: %1").arg(buffer->revision()); Notification n; - n.type = Sink::Commands::NotificationType::NotificationType_RevisionUpdate; + n.type = Sink::Notification::RevisionUpdate; emit notification(n); emit revisionChanged(buffer->revision()); @@ -553,30 +570,26 @@ bool ResourceAccess::processMessageBuffer() case Commands::NotificationCommand: { auto buffer = Commands::GetNotification(d->partialMessageBuffer.constData() + headerSize); switch (buffer->type()) { - case Sink::Commands::NotificationType::NotificationType_Shutdown: + case Sink::Notification::Shutdown: Log() << "Received shutdown notification."; close(); break; - case Sink::Commands::NotificationType::NotificationType_Inspection: { + case Sink::Notification::Inspection: { Trace() << "Received inspection notification."; - Notification n; - if (buffer->identifier()) { - // Don't use fromRawData, the buffer is gone once we invoke emit notification - n.id = BufferUtils::extractBufferCopy(buffer->identifier()); - } - if (buffer->message()) { - // Don't use fromRawData, the buffer is gone once we invoke emit notification - n.message = BufferUtils::extractBufferCopy(buffer->message()); - } - n.type = buffer->type(); - n.code = buffer->code(); + auto n = getNotification(buffer); // The callbacks can result in this object getting destroyed directly, so we need to ensure we finish our work first queuedInvoke([=]() { emit notification(n); }, this); } break; - case Sink::Commands::NotificationType::NotificationType_Status: - case Sink::Commands::NotificationType::NotificationType_Warning: - case Sink::Commands::NotificationType::NotificationType_Progress: - case Sink::Commands::NotificationType::NotificationType_RevisionUpdate: + case Sink::Notification::Status: + mResourceStatus = buffer->code(); + [[clang::fallthrough]]; + case Sink::Notification::Warning: + [[clang::fallthrough]]; + case Sink::Notification::Progress: { + auto n = getNotification(buffer); + emit notification(n); + } break; + case Sink::Notification::RevisionUpdate: default: Warning() << "Received unknown notification: " << buffer->type(); break; diff --git a/common/resourceaccess.h b/common/resourceaccess.h index 69d52b4..5c65998 100644 --- a/common/resourceaccess.h +++ b/common/resourceaccess.h @@ -72,14 +72,22 @@ public: return KAsync::null(); }; + int getResourceStatus() const + { + return mResourceStatus; + } + signals: void ready(bool isReady); void revisionChanged(qint64 revision); - void notification(Notification revision); + void notification(Notification notification); public slots: virtual void open() = 0; virtual void close() = 0; + +protected: + int mResourceStatus; }; class SINK_EXPORT ResourceAccess : public ResourceAccessInterface diff --git a/tests/dummyresourcebenchmark.cpp b/tests/dummyresourcebenchmark.cpp index d5f98c3..7e334a6 100644 --- a/tests/dummyresourcebenchmark.cpp +++ b/tests/dummyresourcebenchmark.cpp @@ -90,7 +90,7 @@ private slots: bool gotNotification = false; int duration = 0; notifier->registerHandler([&gotNotification, &duration, &time](const Sink::Notification ¬ification) { - if (notification.type == Sink::Commands::NotificationType::NotificationType_RevisionUpdate) { + if (notification.type == Sink::Notification::RevisionUpdate) { gotNotification = true; duration = time.elapsed(); } -- cgit v1.2.3