[IndexedDB] Propogating Changes to Observer : Browser

content/browser/indexed_db/indexed_db_transaction.cc

 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "content/browser/indexed_db/indexed_db_transaction.h"
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
+#include "base/memory/ptr_util.h"
 #include "base/single_thread_task_runner.h"
+#include "base/stl_util.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "content/browser/indexed_db/indexed_db_backing_store.h"
 #include "content/browser/indexed_db/indexed_db_cursor.h"
 #include "content/browser/indexed_db/indexed_db_database.h"
 #include "content/browser/indexed_db/indexed_db_database_callbacks.h"
+#include "content/browser/indexed_db/indexed_db_observation.h"
+#include "content/browser/indexed_db/indexed_db_observer_changes.h"
 #include "content/browser/indexed_db/indexed_db_tracing.h"
 #include "content/browser/indexed_db/indexed_db_transaction_coordinator.h"
 #include "third_party/WebKit/public/platform/modules/indexeddb/WebIDBDatabaseException.h"
 #include "third_party/leveldatabase/env_chromium.h"
 
 namespace content {
 
 namespace {
 
 const int64_t kInactivityTimeoutPeriodSeconds = 60;
@@ skipping 32 matching lines @@
 
 IndexedDBTransaction::Operation IndexedDBTransaction::TaskStack::pop() {
   DCHECK(!stack_.empty());
   Operation task(stack_.top());
   stack_.pop();
   return task;
 }
 
 IndexedDBTransaction::IndexedDBTransaction(
     int64_t id,
-    scoped_refptr<IndexedDBDatabaseCallbacks> callbacks,
+    base::WeakPtr<IndexedDBConnection> connection,
     const std::set<int64_t>& object_store_ids,
     blink::WebIDBTransactionMode mode,
-    IndexedDBDatabase* database,
     IndexedDBBackingStore::Transaction* backing_store_transaction)
     : id_(id),
       object_store_ids_(object_store_ids),
       mode_(mode),
       used_(false),
       state_(CREATED),
       commit_pending_(false),
-      callbacks_(callbacks),
-      database_(database),
+      connection_(std::move(connection)),
       transaction_(backing_store_transaction),
       backing_store_transaction_begun_(false),
       should_process_queue_(false),
       pending_preemptive_events_(0) {
+  callbacks_ = connection_->callbacks();
+  database_ = connection_->database();
+
   database_->transaction_coordinator().DidCreateTransaction(this);
 
   diagnostics_.tasks_scheduled = 0;
   diagnostics_.tasks_completed = 0;
   diagnostics_.creation_time = base::Time::Now();
 }
 
 IndexedDBTransaction::~IndexedDBTransaction() {
   // It shouldn't be possible for this object to get deleted until it's either
   // complete or aborted.
   DCHECK_EQ(state_, FINISHED);
   DCHECK(preemptive_task_queue_.empty());
   DCHECK_EQ(pending_preemptive_events_, 0);
   DCHECK(task_queue_.empty());
   DCHECK(abort_task_stack_.empty());
+  DCHECK(pending_observers_.empty());
 }
 
 void IndexedDBTransaction::ScheduleTask(blink::WebIDBTaskType type,
                                         Operation task) {
   DCHECK_NE(state_, COMMITTING);
   if (state_ == FINISHED)
     return;
 
   timeout_timer_.Stop();
   used_ = true;
@@ skipping 73 matching lines @@
 #ifndef NDEBUG
   DCHECK(!database_->transaction_coordinator().IsActive(this));
 #endif
 
   if (callbacks_.get())
     callbacks_->OnAbort(id_, error);
 
   database_->TransactionFinished(this, false);
 
   database_ = NULL;
+  connection_ = nullptr;
+  pending_observers_.clear();
 }
 
 bool IndexedDBTransaction::IsTaskQueueEmpty() const {
   return preemptive_task_queue_.empty() && task_queue_.empty();
 }
 
 bool IndexedDBTransaction::HasPendingTasks() const {
   return pending_preemptive_events_ || !IsTaskQueueEmpty();
 }
 
@@ skipping 129 matching lines @@
   CloseOpenCursors();
   transaction_->Reset();
 
   // Transactions must also be marked as completed before the
   // front-end is notified, as the transaction completion unblocks
   // operations like closing connections.
   database_->transaction_coordinator().DidFinishTransaction(this);
 
   if (committed) {
     abort_task_stack_.clear();
+
+    // SendObservations must be called before OnComplete to ensure consistency
+    // of callbacks at renderer.
+    if (!connection_changes_map_.empty()) {
+      database_->SendObservations(std::move(connection_changes_map_));
+      connection_changes_map_.clear();
+    }
     {
       IDB_TRACE1(
           "IndexedDBTransaction::CommitPhaseTwo.TransactionCompleteCallbacks",
           "txn.id", id());
       callbacks_->OnComplete(id_);
     }
+    if (!pending_observers_.empty() && connection_) {
+      connection_->ActivatePendingObservers(std::move(pending_observers_));
+      pending_observers_.clear();
+    }
+
     database_->TransactionFinished(this, true);
   } else {
     while (!abort_task_stack_.empty())
       abort_task_stack_.pop().Run(NULL);
 
     IndexedDBDatabaseError error;
     if (leveldb_env::IndicatesDiskFull(s)) {
       error = IndexedDBDatabaseError(
           blink::WebIDBDatabaseExceptionQuotaError,
           "Encountered disk full while committing transaction.");
@@ skipping 79 matching lines @@
       base::ASCIIToUTF16("Transaction timed out due to inactivity.")));
 }
 
 void IndexedDBTransaction::CloseOpenCursors() {
   IDB_TRACE1("IndexedDBTransaction::CloseOpenCursors", "txn.id", id());
   for (auto* cursor : open_cursors_)
     cursor->Close();
   open_cursors_.clear();
 }
 
+void IndexedDBTransaction::AddPendingObserver(
+    int32_t observer_id,
+    IndexedDBObserver::Options options) {
+  pending_observers_.push_back(base::WrapUnique(
+      new IndexedDBObserver(observer_id, object_store_ids_, options)));
+}
+
+void IndexedDBTransaction::RemovePendingObservers(
+    const std::vector<int32_t>& pending_observer_ids) {
+  const auto& it = std::remove_if(
+      pending_observers_.begin(), pending_observers_.end(),
+      [&pending_observer_ids](const std::unique_ptr<IndexedDBObserver>& o) {
+        return ContainsValue(pending_observer_ids, o->id());
+      });
+  if (it != pending_observers_.end())
+    pending_observers_.erase(it, pending_observers_.end());
+}
+
+void IndexedDBTransaction::AddObservation(
+    int32_t connection_id,
+    std::unique_ptr<IndexedDBObservation> observation) {
+  auto it = connection_changes_map_.find(connection_id);
+  if (it == connection_changes_map_.end()) {
+    it = connection_changes_map_
+             .insert(std::make_pair(
+                 connection_id,
+                 base::WrapUnique(new IndexedDBObserverChanges())))
+             .first;
+  }
+  it->second->AddObservation(std::move(observation));
+}
+
+void IndexedDBTransaction::RecordObserverForLastObservation(
+    int32_t connection_id,
+    int32_t observer_id) {
+  auto it = connection_changes_map_.find(connection_id);
+  DCHECK(it != connection_changes_map_.end());
+  it->second->RecordObserverForLastObservation(observer_id);
+}
+
 }  // namespace content