Add transaction support. (#199)
Add transaction support to Firebase Realtime Database
diff --git a/packages/firebase_database/CHANGELOG.md b/packages/firebase_database/CHANGELOG.md
index 52188db..2927188 100644
--- a/packages/firebase_database/CHANGELOG.md
+++ b/packages/firebase_database/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.1.1
+
+* Add RTDB transaction support.
+
## 0.1.0+1
* Aligned author name with rest of repo.
diff --git a/packages/firebase_database/android/src/main/java/io/flutter/plugins/firebase/database/FirebaseDatabasePlugin.java b/packages/firebase_database/android/src/main/java/io/flutter/plugins/firebase/database/FirebaseDatabasePlugin.java
index ed9e561..1ac3c05 100644
--- a/packages/firebase_database/android/src/main/java/io/flutter/plugins/firebase/database/FirebaseDatabasePlugin.java
+++ b/packages/firebase_database/android/src/main/java/io/flutter/plugins/firebase/database/FirebaseDatabasePlugin.java
@@ -4,14 +4,20 @@
package io.flutter.plugins.firebase.database;
+import android.util.Log;
import android.util.SparseArray;
+import com.google.android.gms.tasks.Task;
+import com.google.android.gms.tasks.TaskCompletionSource;
+import com.google.android.gms.tasks.Tasks;
import com.google.firebase.database.ChildEventListener;
import com.google.firebase.database.DataSnapshot;
import com.google.firebase.database.DatabaseError;
import com.google.firebase.database.DatabaseException;
import com.google.firebase.database.DatabaseReference;
import com.google.firebase.database.FirebaseDatabase;
+import com.google.firebase.database.MutableData;
import com.google.firebase.database.Query;
+import com.google.firebase.database.Transaction;
import com.google.firebase.database.ValueEventListener;
import io.flutter.plugin.common.MethodCall;
import io.flutter.plugin.common.MethodChannel;
@@ -20,10 +26,15 @@
import io.flutter.plugin.common.PluginRegistry;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
/** FirebaseDatabasePlugin */
public class FirebaseDatabasePlugin implements MethodCallHandler {
+ private static final String TAG = "FirebaseDatabasePlugin";
+
private final MethodChannel channel;
private static final String EVENT_TYPE_CHILD_ADDED = "_EventType.childAdded";
private static final String EVENT_TYPE_CHILD_REMOVED = "_EventType.childRemoved";
@@ -197,7 +208,7 @@
}
@Override
- public void onMethodCall(MethodCall call, final Result result) {
+ public void onMethodCall(final MethodCall call, final Result result) {
switch (call.method) {
case "FirebaseDatabase#goOnline":
{
@@ -278,6 +289,108 @@
break;
}
+ case "DatabaseReference#runTransaction":
+ {
+ final Map<String, Object> arguments = call.arguments();
+ final DatabaseReference reference = getReference(arguments);
+
+ // Initiate native transaction.
+ reference.runTransaction(
+ new Transaction.Handler() {
+ @Override
+ public Transaction.Result doTransaction(MutableData mutableData) {
+ // Tasks are used to allow native execution of doTransaction to wait while Snapshot is
+ // processed by logic on the Dart side.
+ final TaskCompletionSource<Map<String, Object>> updateMutableDataTCS =
+ new TaskCompletionSource<>();
+ final Task<Map<String, Object>> updateMutableDataTCSTask =
+ updateMutableDataTCS.getTask();
+
+ Map<String, Object> doTransactionMap = new HashMap<>();
+ doTransactionMap.put("transactionKey", arguments.get("transactionKey"));
+
+ final Map<String, Object> snapshotMap = new HashMap<>();
+ snapshotMap.put("key", mutableData.getKey());
+ snapshotMap.put("value", mutableData.getValue());
+ doTransactionMap.put("snapshot", snapshotMap);
+
+ // Return snapshot to Dart side for update.
+ channel.invokeMethod(
+ "DoTransaction",
+ doTransactionMap,
+ new MethodChannel.Result() {
+ @Override
+ @SuppressWarnings("unchecked")
+ public void success(Object result) {
+ updateMutableDataTCS.setResult((Map<String, Object>) result);
+ }
+
+ @Override
+ public void error(
+ String errorCode, String errorMessage, Object errorDetails) {
+ String exceptionMessage =
+ "Error code: "
+ + errorCode
+ + "\nError message: "
+ + errorMessage
+ + "\nError details: "
+ + errorDetails;
+ updateMutableDataTCS.setException(new Exception(exceptionMessage));
+ }
+
+ @Override
+ public void notImplemented() {
+ updateMutableDataTCS.setException(
+ new Exception("DoTransaction not implemented on Dart side."));
+ }
+ });
+
+ try {
+ // Wait for updated snapshot from the Dart side.
+ Map<String, Object> updatedSnapshotMap =
+ Tasks.await(
+ updateMutableDataTCSTask,
+ (int) arguments.get("transactionTimeout"),
+ TimeUnit.MILLISECONDS);
+ // Set value of MutableData to value returned from the Dart side.
+ mutableData.setValue(updatedSnapshotMap.get("value"));
+ } catch (ExecutionException | InterruptedException | TimeoutException e) {
+ Log.e(TAG, "Unable to commit Snapshot update. Transaction failed.", e);
+ if (e instanceof TimeoutException) {
+ Log.e(TAG, "Transaction at " + reference.toString() + " timed out.");
+ }
+ return Transaction.abort();
+ }
+ return Transaction.success(mutableData);
+ }
+
+ @Override
+ public void onComplete(
+ DatabaseError databaseError, boolean committed, DataSnapshot dataSnapshot) {
+ Map<String, Object> completionMap = new HashMap<>();
+ completionMap.put("transactionKey", arguments.get("transactionKey"));
+ if (databaseError != null) {
+ Map<String, Object> errorMap = new HashMap<>();
+ errorMap.put("code", databaseError.getCode());
+ errorMap.put("message", databaseError.getMessage());
+ errorMap.put("details", databaseError.getDetails());
+ completionMap.put("error", errorMap);
+ }
+ completionMap.put("committed", committed);
+ if (dataSnapshot != null) {
+ Map<String, Object> snapshotMap = new HashMap<>();
+ snapshotMap.put("key", dataSnapshot.getKey());
+ snapshotMap.put("value", dataSnapshot.getValue());
+ completionMap.put("snapshot", snapshotMap);
+ }
+
+ // Invoke transaction completion on the Dart side.
+ result.success(completionMap);
+ }
+ });
+ break;
+ }
+
case "Query#keepSynced":
{
Map<String, Object> arguments = call.arguments();
diff --git a/packages/firebase_database/example/lib/main.dart b/packages/firebase_database/example/lib/main.dart
index afce2a2..3ba755c 100755
--- a/packages/firebase_database/example/lib/main.dart
+++ b/packages/firebase_database/example/lib/main.dart
@@ -67,15 +67,23 @@
Future<Null> _increment() async {
await FirebaseAuth.instance.signInAnonymously();
- // TODO(jackson): This illustrates a case where transactions are needed
- final DataSnapshot snapshot = await _counterRef.once();
- setState(() {
- _counter = (snapshot.value ?? 0) + 1;
+ // Increment counter in transaction.
+ final TransactionResult transactionResult =
+ await _counterRef.runTransaction((MutableData mutableData) async {
+ mutableData.value = (mutableData.value ?? 0) + 1;
+ return mutableData;
});
- _counterRef.set(_counter);
- _messagesRef
- .push()
- .set(<String, String>{_kTestKey: '$_kTestValue $_counter'});
+
+ if (transactionResult.committed) {
+ _messagesRef.push().set(<String, String>{
+ _kTestKey: '$_kTestValue ${transactionResult.dataSnapshot.value}'
+ });
+ } else {
+ print('Transaction not committed.');
+ if (transactionResult.error != null) {
+ print(transactionResult.error.message);
+ }
+ }
}
@override
diff --git a/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.h b/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.h
index deff3a4..3a360f9 100644
--- a/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.h
+++ b/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.h
@@ -5,4 +5,7 @@
#import <Flutter/Flutter.h>
@interface FirebaseDatabasePlugin : NSObject<FlutterPlugin>
+
+@property(nonatomic) NSMutableDictionary *updatedSnapshots;
+
@end
diff --git a/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.m b/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.m
index 65a5a23..89966bc 100644
--- a/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.m
+++ b/packages/firebase_database/ios/Classes/FirebaseDatabasePlugin.m
@@ -6,6 +6,9 @@
#import <Firebase/Firebase.h>
+@interface FirebaseDatabasePlugin ()
+@end
+
@interface NSError (FlutterError)
@property(readonly, nonatomic) FlutterError *flutterError;
@end
@@ -136,6 +139,7 @@
if (![FIRApp defaultApp]) {
[FIRApp configure];
}
+ self.updatedSnapshots = [NSMutableDictionary new];
}
return self;
}
@@ -190,6 +194,69 @@
} else if ([@"DatabaseReference#setPriority" isEqualToString:call.method]) {
[getReference(call.arguments) setPriority:call.arguments[@"priority"]
withCompletionBlock:defaultCompletionBlock];
+ } else if ([@"DatabaseReference#runTransaction" isEqualToString:call.method]) {
+ [getReference(call.arguments) runTransactionBlock:^FIRTransactionResult *_Nonnull(
+ FIRMutableData *_Nonnull currentData) {
+ // Create semaphore to allow native side to wait while snapshot
+ // updates occur on the Dart side.
+ dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);
+
+ NSObject *snapshot =
+ @{@"key" : currentData.key ?: [NSNull null], @"value" : currentData.value};
+
+ __block bool shouldAbort = false;
+
+ [self.channel invokeMethod:@"DoTransaction"
+ arguments:@{
+ @"transactionKey" : call.arguments[@"transactionKey"],
+ @"snapshot" : snapshot
+ }
+ result:^(id _Nullable result) {
+ if ([result isKindOfClass:[FlutterError class]]) {
+ FlutterError *flutterError = ((FlutterError *)result);
+ NSLog(@"Error code: %@", flutterError.code);
+ NSLog(@"Error message: %@", flutterError.message);
+ NSLog(@"Error details: %@", flutterError.details);
+ shouldAbort = true;
+ } else if ([result isEqual:FlutterMethodNotImplemented]) {
+ NSLog(@"DoTransaction not implemented on the Dart side.");
+ shouldAbort = true;
+ } else {
+ [self.updatedSnapshots setObject:result
+ forKey:call.arguments[@"transactionKey"]];
+ }
+ dispatch_semaphore_signal(semaphore);
+ }];
+
+ // Wait while Dart side updates the snapshot. Incoming transactionTimeout is in milliseconds
+ // so converting to nanoseconds for use with dispatch_semaphore_wait.
+ long result = dispatch_semaphore_wait(
+ semaphore, dispatch_time(DISPATCH_TIME_NOW,
+ [call.arguments[@"transactionTimeout"] integerValue] * 1000000));
+
+ if (result == 0 && !shouldAbort) {
+ // Set FIRMutableData value to value returned from the Dart side.
+ currentData.value =
+ [self.updatedSnapshots objectForKey:call.arguments[@"transactionKey"]][@"value"];
+ } else {
+ if (result != 0) {
+ NSLog(@"Transaction at %@ timed out.", [getReference(call.arguments) URL]);
+ }
+ return [FIRTransactionResult abort];
+ }
+
+ return [FIRTransactionResult successWithValue:currentData];
+ }
+ andCompletionBlock:^(NSError *_Nullable error, BOOL committed,
+ FIRDataSnapshot *_Nullable snapshot) {
+ // Invoke transaction completion on the Dart side.
+ result(@{
+ @"transactionKey" : call.arguments[@"transactionKey"],
+ @"error" : error.flutterError ?: [NSNull null],
+ @"committed" : [NSNumber numberWithBool:committed],
+ @"snapshot" : @{@"key" : snapshot.key ?: [NSNull null], @"value" : snapshot.value}
+ });
+ }];
} else if ([@"Query#observe" isEqualToString:call.method]) {
FIRDataEventType eventType = parseEventType(call.arguments[@"eventType"]);
__block FIRDatabaseHandle handle = [getQuery(call.arguments)
diff --git a/packages/firebase_database/lib/src/database_reference.dart b/packages/firebase_database/lib/src/database_reference.dart
index a668b07..ff584cc 100644
--- a/packages/firebase_database/lib/src/database_reference.dart
+++ b/packages/firebase_database/lib/src/database_reference.dart
@@ -125,6 +125,45 @@
///
/// remove() is equivalent to calling set(null)
Future<Null> remove() => set(null);
+
+ /// Performs an optimistic-concurrency transactional update to the data at
+ /// this Firebase Database location.
+ Future<TransactionResult> runTransaction(
+ TransactionHandler transactionHandler,
+ {Duration timeout: const Duration(seconds: 5)}) async {
+ assert(timeout.inMilliseconds > 0,
+ 'Transaction timeout must be more than 0 milliseconds.');
+
+ final Completer<TransactionResult> completer =
+ new Completer<TransactionResult>();
+
+ final int transactionKey = FirebaseDatabase._transactions.isEmpty
+ ? 0
+ : FirebaseDatabase._transactions.keys.last + 1;
+
+ FirebaseDatabase._transactions[transactionKey] = transactionHandler;
+
+ _database._channel
+ .invokeMethod('DatabaseReference#runTransaction', <String, dynamic>{
+ 'path': path,
+ 'transactionKey': transactionKey,
+ 'transactionTimeout': timeout.inMilliseconds
+ }).then((Map<String, dynamic> result) {
+ final DatabaseError databaseError =
+ result['error'] != null ? new DatabaseError._(result['error']) : null;
+ final bool committed = result['committed'];
+ final DataSnapshot dataSnapshot = result['snapshot'] != null
+ ? new DataSnapshot._(result['snapshot'])
+ : null;
+
+ FirebaseDatabase._transactions.remove(transactionKey);
+
+ completer.complete(
+ new TransactionResult._(databaseError, committed, dataSnapshot));
+ });
+
+ return completer.future;
+ }
}
class ServerValue {
@@ -132,3 +171,12 @@
'.sv': 'timestamp'
};
}
+
+typedef Future<MutableData> TransactionHandler(MutableData mutableData);
+
+class TransactionResult {
+ const TransactionResult._(this.error, this.committed, this.dataSnapshot);
+ final DatabaseError error;
+ final bool committed;
+ final DataSnapshot dataSnapshot;
+}
diff --git a/packages/firebase_database/lib/src/event.dart b/packages/firebase_database/lib/src/event.dart
index f3b902a..4be4c7c 100644
--- a/packages/firebase_database/lib/src/event.dart
+++ b/packages/firebase_database/lib/src/event.dart
@@ -25,7 +25,7 @@
/// A DataSnapshot contains data from a Firebase Database location.
/// Any time you read Firebase data, you receive the data as a DataSnapshot.
class DataSnapshot {
- Map<String, dynamic> _data;
+ final Map<String, dynamic> _data;
DataSnapshot._(this._data);
/// The key of the location that generated this DataSnapshot.
@@ -34,3 +34,33 @@
/// Returns the contents of this data snapshot as native types.
dynamic get value => _data['value'];
}
+
+class MutableData {
+ final Map<String, dynamic> _data;
+ @visibleForTesting
+ MutableData.private(this._data);
+
+ /// The key of the location that generated this MutableData.
+ String get key => _data['key'];
+
+ /// Returns the mutable contents of this MutableData as native types.
+ dynamic get value => _data['value'];
+ set value(dynamic newValue) => _data['value'] = newValue;
+}
+
+/// A DatabaseError contains code, message and details of a Firebase Database
+/// Error that results from a transaction operation at a Firebase Database
+/// location.
+class DatabaseError {
+ Map<String, dynamic> _data;
+ DatabaseError._(this._data);
+
+ /// One of the defined status codes, depending on the error.
+ int get code => _data['code'];
+
+ /// A human-readable description of the error.
+ String get message => _data['message'];
+
+ /// Human-readable details on the error and additional information.
+ String get details => _data['details'];
+}
diff --git a/packages/firebase_database/lib/src/firebase_database.dart b/packages/firebase_database/lib/src/firebase_database.dart
index 678f0dd..4ab151e 100644
--- a/packages/firebase_database/lib/src/firebase_database.dart
+++ b/packages/firebase_database/lib/src/firebase_database.dart
@@ -15,11 +15,24 @@
static final Map<int, StreamController<Event>> _observers =
<int, StreamController<Event>>{};
+ static final Map<int, TransactionHandler> _transactions =
+ <int, TransactionHandler>{};
+
FirebaseDatabase._() {
- _channel.setMethodCallHandler((MethodCall call) {
+ _channel.setMethodCallHandler((MethodCall call) async {
if (call.method == 'Event') {
final Event event = new Event._(call.arguments);
_observers[call.arguments['handle']].add(event);
+ } else if (call.method == 'DoTransaction') {
+ final MutableData mutableData =
+ new MutableData.private(call.arguments['snapshot']);
+ final MutableData updated =
+ await _transactions[call.arguments['transactionKey']](mutableData);
+ return <String, dynamic>{'value': updated.value};
+ } else {
+ throw new MissingPluginException(
+ '${call.method} method not implemented on the Dart side.',
+ );
}
});
}
diff --git a/packages/firebase_database/pubspec.yaml b/packages/firebase_database/pubspec.yaml
index 6d8aac1..f6e61a0 100755
--- a/packages/firebase_database/pubspec.yaml
+++ b/packages/firebase_database/pubspec.yaml
@@ -3,7 +3,7 @@
description: Firebase Database plugin for Flutter.
author: Flutter Team <flutter-dev@googlegroups.com>
homepage: https://github.com/flutter/plugins/tree/master/packages/firebase_database
-version: 0.1.0+1
+version: 0.1.1
flutter:
plugin:
diff --git a/packages/firebase_database/test/firebase_database_test.dart b/packages/firebase_database/test/firebase_database_test.dart
index 27a329d..24a87e7 100755
--- a/packages/firebase_database/test/firebase_database_test.dart
+++ b/packages/firebase_database/test/firebase_database_test.dart
@@ -28,6 +28,41 @@
return true;
case 'FirebaseDatabase#setPersistenceCacheSizeBytes':
return true;
+ case 'DatabaseReference#runTransaction':
+ Map<String, dynamic> updatedValue;
+ Future<Null> simulateEvent(
+ int transactionKey, final MutableData mutableData) async {
+ await BinaryMessages.handlePlatformMessage(
+ channel.name,
+ channel.codec.encodeMethodCall(
+ new MethodCall('DoTransaction', <String, dynamic>{
+ 'transactionKey': transactionKey,
+ 'snapshot': <String, dynamic>{
+ 'key': mutableData.key,
+ 'value': mutableData.value,
+ },
+ })),
+ (_) {
+ updatedValue = channel.codec.decodeEnvelope(_)['value'];
+ },
+ );
+ }
+
+ await simulateEvent(
+ 0,
+ new MutableData.private(<String, dynamic>{
+ 'key': 'fakeKey',
+ 'value': <String, dynamic>{'fakeKey': 'fakeValue'},
+ }));
+
+ return <String, dynamic>{
+ 'error': null,
+ 'committed': true,
+ 'snapshot': <String, dynamic>{
+ 'key': 'fakeKey',
+ 'value': updatedValue,
+ }
+ };
default:
return null;
}
@@ -142,6 +177,40 @@
]),
);
});
+
+ test('runTransaction', () async {
+ final TransactionResult transactionResult = await database
+ .reference()
+ .child('foo')
+ .runTransaction((MutableData mutableData) {
+ return new Future<MutableData>(() {
+ mutableData.value['fakeKey'] =
+ 'updated ' + mutableData.value['fakeKey'];
+ return mutableData;
+ });
+ });
+ expect(
+ log,
+ equals(<MethodCall>[
+ new MethodCall(
+ 'DatabaseReference#runTransaction', <String, dynamic>{
+ 'path': 'foo',
+ 'transactionKey': 0,
+ 'transactionTimeout': 5000
+ }),
+ ]),
+ );
+ expect(transactionResult.committed, equals(true));
+ expect(transactionResult.dataSnapshot.value,
+ equals(<String, dynamic>{'fakeKey': 'updated fakeValue'}));
+ expect(
+ database.reference().child('foo').runTransaction(
+ (MutableData mutableData) {},
+ timeout: const Duration(milliseconds: 0),
+ ),
+ throwsA(const isInstanceOf<AssertionError>()),
+ );
+ });
});
group('$Query', () {
