diff options
author | Dan Vrátil <dvratil@redhat.com> | 2015-02-21 12:11:42 +0100 |
---|---|---|
committer | Dan Vrátil <dvratil@redhat.com> | 2015-02-21 12:11:44 +0100 |
commit | 3b8ebe6d4235f5ba12bc9c9854a6dd28cbff06b5 (patch) | |
tree | 4e2a987e6c62523994965789387cace578e4c8d8 /common/resourceaccess.cpp | |
parent | 76ec0cfe075e3af758657f9aecab7d7ce7e8d387 (diff) | |
download | sink-3b8ebe6d4235f5ba12bc9c9854a6dd28cbff06b5.tar.gz sink-3b8ebe6d4235f5ba12bc9c9854a6dd28cbff06b5.zip |
Async: allow appending Jobs to already running or finished Jobs
When user gets a Job (from a method call for instance), which is already running
or might have even finished already, they can still append a new Job to the chain
and re-execute it. The Job will internally chain up to the last finished Job, use
it's result and continue from the next Job in the chain. If a Job in the chain is
still running, it will wait for it to finish and pass the result to the next Job
in the chain.
Diffstat (limited to 'common/resourceaccess.cpp')
-rw-r--r-- | common/resourceaccess.cpp | 57 |
1 files changed, 15 insertions, 42 deletions
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp index c806478..ffe716b 100644 --- a/common/resourceaccess.cpp +++ b/common/resourceaccess.cpp | |||
@@ -154,51 +154,24 @@ Async::Job<void> ResourceAccess::sendCommand(int commandId) | |||
154 | }); | 154 | }); |
155 | } | 155 | } |
156 | 156 | ||
157 | /* | ||
158 | * TODO JOBAPI: This is a workaround to be able to return a job below to | ||
159 | * may or may not already be finished when the job is started. The job API should provide a mechanism | ||
160 | * for this. Essentially we need a way to set a job finished externally (we use the finisher as handle for that). | ||
161 | * If the job is then started the continuation should immediately be executed if the job finished already, and otherwise | ||
162 | * just wait until the work is done, and then execute the continuation as usual. | ||
163 | */ | ||
164 | struct JobFinisher { | ||
165 | bool finished; | ||
166 | std::function<void(int error, const QString &errorMessage)> callback; | ||
167 | |||
168 | JobFinisher() : finished(false) {} | ||
169 | |||
170 | void setFinished(int error, const QString &errorMessage) { | ||
171 | finished = true; | ||
172 | if (callback) { | ||
173 | callback(error, errorMessage); | ||
174 | } | ||
175 | } | ||
176 | }; | ||
177 | |||
178 | Async::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) | 157 | Async::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) |
179 | { | 158 | { |
180 | auto finisher = QSharedPointer<JobFinisher>::create(); | 159 | return Async::start<void>([commandId, &fbb, this](Async::Future<void> &f) { |
181 | auto callback = [finisher] (int error, const QString &errorMessage) { | 160 | auto callback = [&f](int error, const QString &errorMessage) { |
182 | finisher->setFinished(error, errorMessage); | 161 | if (error) { |
183 | }; | 162 | f.setError(error, errorMessage); |
184 | if (isReady()) { | 163 | } else { |
185 | d->messageId++; | ||
186 | log(QString("Sending command %1 with messageId %2").arg(commandId).arg(d->messageId)); | ||
187 | registerCallback(d->messageId, callback); | ||
188 | Commands::write(d->socket, d->messageId, commandId, fbb); | ||
189 | } else { | ||
190 | d->commandQueue << new QueuedCommand(commandId, fbb, callback); | ||
191 | } | ||
192 | return Async::start<void>([this, finisher](Async::Future<void> &f) { | ||
193 | if (finisher->finished) { | ||
194 | f.setFinished(); | ||
195 | } else { | ||
196 | finisher->callback = [&f](int error, const QString &errorMessage) { | ||
197 | if (error) { | ||
198 | f.setError(error, errorMessage); | ||
199 | } | ||
200 | f.setFinished(); | 164 | f.setFinished(); |
201 | }; | 165 | } |
166 | }; | ||
167 | |||
168 | if (isReady()) { | ||
169 | d->messageId++; | ||
170 | log(QString("Sending command %1 with messageId %2").arg(commandId).arg(d->messageId)); | ||
171 | registerCallback(d->messageId, callback); | ||
172 | Commands::write(d->socket, d->messageId, commandId, fbb); | ||
173 | } else { | ||
174 | d->commandQueue << new QueuedCommand(commandId, fbb, callback); | ||
202 | } | 175 | } |
203 | }); | 176 | }); |
204 | } | 177 | } |