buffet: Order device state and command updates on the server
Added strict ordering between command and state updates on the cloud
server interface. Command updates are tied to the current device state
and command update requests to the server are not dispatched until the
corresponding device state request finishes successfully.
BUG=brillo:1202
TEST=`FEATURES=test emerge-link buffet`
Change-Id: I23af95ab66b5bca91f637d9886ae234681b67104
Reviewed-on: https://chromium-review.googlesource.com/282261
Tested-by: Alex Vakulenko <avakulenko@chromium.org>
Reviewed-by: Vitaly Buka <vitalybuka@chromium.org>
Commit-Queue: Alex Vakulenko <avakulenko@chromium.org>
diff --git a/buffet/commands/cloud_command_proxy.h b/buffet/commands/cloud_command_proxy.h
index 2c607ea..55eb552 100644
--- a/buffet/commands/cloud_command_proxy.h
+++ b/buffet/commands/cloud_command_proxy.h
@@ -5,14 +5,20 @@
#ifndef BUFFET_COMMANDS_CLOUD_COMMAND_PROXY_H_
#define BUFFET_COMMANDS_CLOUD_COMMAND_PROXY_H_
-#include <bitset>
+#include <deque>
+#include <memory>
#include <string>
+#include <utility>
#include <base/macros.h>
+#include <base/memory/ref_counted.h>
#include <base/memory/weak_ptr.h>
+#include <base/task_runner.h>
+#include <chromeos/backoff_entry.h>
#include "buffet/commands/cloud_command_update_interface.h"
#include "buffet/commands/command_proxy_interface.h"
+#include "buffet/states/state_change_queue_interface.h"
namespace buffet {
@@ -22,7 +28,10 @@
class CloudCommandProxy final : public CommandProxyInterface {
public:
CloudCommandProxy(CommandInstance* command_instance,
- CloudCommandUpdateInterface* cloud_command_updater);
+ CloudCommandUpdateInterface* cloud_command_updater,
+ StateChangeQueueInterface* state_change_queue,
+ std::unique_ptr<chromeos::BackoffEntry> backoff_entry,
+ const scoped_refptr<base::TaskRunner>& task_runner);
~CloudCommandProxy() override = default;
// CommandProxyInterface implementation/overloads.
@@ -31,14 +40,20 @@
void OnProgressChanged() override;
private:
- // Flags used to mark the command resource parts that need to be updated on
- // the server.
- using CommandUpdateFlags = std::bitset<3>;
+ using UpdateID = StateChangeQueueInterface::UpdateID;
+ using UpdateQueueEntry =
+ std::pair<UpdateID, std::unique_ptr<base::DictionaryValue>>;
- // Sends an asynchronous request to GCD server to update the command resource.
+ // Puts a command update data into the update queue, and optionally sends an
+ // asynchronous request to GCD server to update the command resource, if there
+ // are no pending device status updates.
+ void QueueCommandUpdate(std::unique_ptr<base::DictionaryValue> patch);
+
+ // Sends an asynchronous request to GCD server to update the command resource,
+ // if there are no pending device status updates.
void SendCommandUpdate();
- // Retry last failed request.
+ // Retry the last failed command update request to the server.
void ResendCommandUpdate();
// Callback invoked by the asynchronous PATCH request to the server.
@@ -46,16 +61,35 @@
// and in case of an error, indicated by the |success| parameter.
void OnUpdateCommandFinished(bool success);
+ // Callback invoked by the device state change queue to notify of the
+ // successful device state update. |update_id| is the ID of the state that
+ // has been updated on the server.
+ void OnDeviceStateUpdated(UpdateID update_id);
+
CommandInstance* command_instance_;
CloudCommandUpdateInterface* cloud_command_updater_;
+ StateChangeQueueInterface* state_change_queue_;
+ scoped_refptr<base::TaskRunner> task_runner_;
+
+ // Backoff for SendCommandUpdate() method.
+ std::unique_ptr<chromeos::BackoffEntry> cloud_backoff_entry_;
// Set to true while a pending PATCH request is in flight to the server.
bool command_update_in_progress_{false};
- // The flags indicating of new command resource updates since the last req.
- CommandUpdateFlags new_pending_command_updates_;
- // The flags indicating of command updates currently in flight.
- CommandUpdateFlags in_progress_command_updates_;
+ // Update queue with all the command update requests ready to be sent to
+ // the server.
+ std::deque<UpdateQueueEntry> update_queue_;
+ // Callback token from the state change queue for OnDeviceStateUpdated()
+ // callback for ask the device state change queue to call when the state
+ // is updated on the server.
+ StateChangeQueueInterface::Token callback_token_;
+
+ // Last device state update ID that has been sent out to the server
+ // successfully.
+ UpdateID last_state_update_id_{0};
+
+ base::WeakPtrFactory<CloudCommandProxy> backoff_weak_ptr_factory_{this};
base::WeakPtrFactory<CloudCommandProxy> weak_ptr_factory_{this};
DISALLOW_COPY_AND_ASSIGN(CloudCommandProxy);
};
