Add onCommitted listeners for transactions (#2265)

Author: schmidt-sebastian

packages/firestore/src/local/indexeddb_persistence.ts

@@ -749,12 +749,17 @@ export class IndexedDbPersistence implements Persistence {
       ? 'readwrite-idempotent'
       : 'readwrite';
 
+    let persistenceTransaction: PersistenceTransaction;
+
     // Do all transactions as readwrite against all object stores, since we
     // are the only reader/writer.
-    return this.simpleDb.runTransaction(
-      simpleDbMode,
-      ALL_STORES,
-      simpleDbTxn => {
+    return this.simpleDb
+      .runTransaction(simpleDbMode, ALL_STORES, simpleDbTxn => {
+        persistenceTransaction = new IndexedDbTransaction(
+          simpleDbTxn,
+          this.listenSequence.next()
+        );
+
         if (mode === 'readwrite-primary') {
           // While we merely verify that we have (or can acquire) the lease
           // immediately, we wait to extend the primary lease until after
@@ -776,12 +781,7 @@ export class IndexedDbPersistence implements Persistence {
                   PRIMARY_LEASE_LOST_ERROR_MSG
                 );
               }
-              return transactionOperation(
-                new IndexedDbTransaction(
-                  simpleDbTxn,
-                  this.listenSequence.next()
-                )
-              );
+              return transactionOperation(persistenceTransaction);
             })
             .next(result => {
               return this.acquireOrExtendPrimaryLease(simpleDbTxn).next(
@@ -790,13 +790,14 @@ export class IndexedDbPersistence implements Persistence {
             });
         } else {
           return this.verifyAllowTabSynchronization(simpleDbTxn).next(() =>
-            transactionOperation(
-              new IndexedDbTransaction(simpleDbTxn, this.listenSequence.next())
-            )
+            transactionOperation(persistenceTransaction)
           );
         }
-      }
-    );
+      })
+      .then(result => {
+        persistenceTransaction.raiseOnCommittedEvent();
+        return result;
+      });
   }
 
   /**

packages/firestore/src/local/memory_persistence.ts

@@ -184,7 +184,11 @@ export class MemoryPersistence implements Persistence {
           .onTransactionCommitted(txn)
           .next(() => result);
       })
-      .toPromise();
+      .toPromise()
+      .then(result => {
+        txn.raiseOnCommittedEvent();
+        return result;
+      });
   }
 
   mutationQueuesContainKey(
@@ -203,8 +207,10 @@ export class MemoryPersistence implements Persistence {
  * Memory persistence is not actually transactional, but future implementations
  * may have transaction-scoped state.
  */
-export class MemoryTransaction implements PersistenceTransaction {
-  constructor(readonly currentSequenceNumber: ListenSequenceNumber) {}
+export class MemoryTransaction extends PersistenceTransaction {
+  constructor(readonly currentSequenceNumber: ListenSequenceNumber) {
+    super();
+  }
 }
 
 export class MemoryEagerDelegate implements ReferenceDelegate {

packages/firestore/src/local/persistence.ts

@@ -29,14 +29,25 @@ import { RemoteDocumentCache } from './remote_document_cache';
 import { ClientId } from './shared_client_state';
 
 /**
- * Opaque interface representing a persistence transaction.
+ * A base class representing a persistence transaction, encapsulating both the
+ * transaction's sequence numbers as well as a list of onCommitted listeners.
  *
  * When you call Persistence.runTransaction(), it will create a transaction and
  * pass it to your callback. You then pass it to any method that operates
  * on persistence.
  */
 export abstract class PersistenceTransaction {
+  private readonly onCommittedListeners: Array<() => void> = [];
+
   abstract readonly currentSequenceNumber: ListenSequenceNumber;
+
+  addOnCommittedListener(listener: () => void): void {
+    this.onCommittedListeners.push(listener);
+  }
+
+  raiseOnCommittedEvent(): void {
+    this.onCommittedListeners.forEach(listener => listener());
+  }
 }
 
 /** The different modes supported by `IndexedDbPersistence.runTransaction()`. */

packages/firestore/test/unit/local/persistence_transaction.test.ts

@@ -0,0 +1,122 @@
+/**
+ * @license
+ * Copyright 2019 Google Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as persistenceHelpers from './persistence_test_helpers';
+import { expect } from 'chai';
+import { IndexedDbPersistence } from '../../../src/local/indexeddb_persistence';
+import { Persistence } from '../../../src/local/persistence';
+import { PersistencePromise } from '../../../src/local/persistence_promise';
+import { DbTarget, DbTargetKey } from '../../../src/local/indexeddb_schema';
+import { TargetId } from '../../../src/core/types';
+
+let persistence: Persistence;
+
+describe('MemoryTransaction', () => {
+  beforeEach(() => {
+    return persistenceHelpers.testMemoryEagerPersistence().then(p => {
+      persistence = p;
+    });
+  });
+
+  genericTransactionTests();
+});
+
+describe('IndexedDbTransaction', () => {
+  if (!IndexedDbPersistence.isAvailable()) {
+    console.warn('No IndexedDB. Skipping IndexedDbTransaction tests.');
+    return;
+  }
+
+  beforeEach(() => {
+    return persistenceHelpers.testIndexedDbPersistence().then(p => {
+      persistence = p;
+    });
+  });
+
+  afterEach(() => persistence.shutdown());
+
+  genericTransactionTests();
+
+  it('only invokes onCommittedListener once with retries', async () => {
+    let runCount = 0;
+    let commitCount = 0;
+    await persistence.runTransaction(
+      'onCommitted',
+      'readwrite-idempotent',
+      txn => {
+        const targetsStore = IndexedDbPersistence.getStore<
+          DbTargetKey,
+          { targetId: TargetId }
+        >(txn, DbTarget.store);
+
+        txn.addOnCommittedListener(() => {
+          ++commitCount;
+        });
+
+        expect(commitCount).to.equal(0);
+
+        ++runCount;
+        if (runCount === 1) {
+          // Trigger a unique key violation
+          return targetsStore
+            .add({ targetId: 1 })
+            .next(() => targetsStore.add({ targetId: 1 }));
+        } else {
+          return PersistencePromise.resolve(0);
+        }
+      }
+    );
+
+    expect(runCount).to.be.equal(2);
+    expect(commitCount).to.be.equal(1);
+  });
+});
+
+function genericTransactionTests(): void {
+  it('invokes onCommittedListener when transaction succeeds', async () => {
+    let onCommitted = false;
+    await persistence.runTransaction(
+      'onCommitted',
+      'readonly-idempotent',
+      txn => {
+        txn.addOnCommittedListener(() => {
+          onCommitted = true;
+        });
+
+        expect(onCommitted).to.be.false;
+        return PersistencePromise.resolve();
+      }
+    );
+
+    expect(onCommitted).to.be.true;
+  });
+
+  it('does not invoke onCommittedListener when transaction fails', async () => {
+    let onCommitted = false;
+    await persistence
+      .runTransaction('onCommitted', 'readonly-idempotent', txn => {
+        txn.addOnCommittedListener(() => {
+          onCommitted = true;
+        });
+
+        return PersistencePromise.reject(new Error('Aborted'));
+      })
+      .catch(() => {});
+
+    expect(onCommitted).to.be.false;
+  });
+}