| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119 |
- /*
- * Copyright 2017 Google
- *
- * 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 <Foundation/Foundation.h>
- #import <FirebaseCore/FIRLogger.h>
- #import <dlfcn.h>
- #import "FRepo.h"
- #import "FSnapshotUtilities.h"
- #import "FConstants.h"
- #import "FIRDatabaseQuery_Private.h"
- #import "FQuerySpec.h"
- #import "FTupleNodePath.h"
- #import "FRepo_Private.h"
- #import "FRepoManager.h"
- #import "FServerValues.h"
- #import "FTupleSetIdPath.h"
- #import "FSyncTree.h"
- #import "FEventRegistration.h"
- #import "FAtomicNumber.h"
- #import "FSyncTree.h"
- #import "FListenProvider.h"
- #import "FEventRaiser.h"
- #import "FSnapshotHolder.h"
- #import "FIRDatabaseConfig_Private.h"
- #import "FLevelDBStorageEngine.h"
- #import "FPersistenceManager.h"
- #import "FWriteRecord.h"
- #import "FCachePolicy.h"
- #import "FClock.h"
- #import "FIRDatabase_Private.h"
- #import "FTree.h"
- #import "FTupleTransaction.h"
- #import "FIRTransactionResult.h"
- #import "FIRTransactionResult_Private.h"
- #import "FIRMutableData.h"
- #import "FIRMutableData_Private.h"
- #import "FIRDataSnapshot.h"
- #import "FIRDataSnapshot_Private.h"
- #import "FValueEventRegistration.h"
- #import "FEmptyNode.h"
- #if TARGET_OS_IOS || TARGET_OS_TV
- #import <UIKit/UIKit.h>
- #endif
- @interface FRepo()
- @property (nonatomic, strong) FOffsetClock *serverClock;
- @property (nonatomic, strong) FPersistenceManager* persistenceManager;
- @property (nonatomic, strong) FIRDatabase *database;
- @property (nonatomic, strong, readwrite) FAuthenticationManager *auth;
- @property (nonatomic, strong) FSyncTree *infoSyncTree;
- @property (nonatomic) NSInteger writeIdCounter;
- @property (nonatomic) BOOL hijackHash;
- @property (nonatomic, strong) FTree *transactionQueueTree;
- @property (nonatomic) BOOL loggedTransactionPersistenceWarning;
- /**
- * Test only. For load testing the server.
- */
- @property (nonatomic, strong) id (^interceptServerDataCallback)(NSString *pathString, id data);
- @end
- @implementation FRepo
- - (id)initWithRepoInfo:(FRepoInfo*)info config:(FIRDatabaseConfig *)config database:(FIRDatabase *)database {
- self = [super init];
- if (self) {
- self.repoInfo = info;
- self.config = config;
- self.database = database;
- // Access can occur outside of shared queue, so the clock needs to be initialized here
- self.serverClock = [[FOffsetClock alloc] initWithClock:[FSystemClock clock] offset:0];
- self.connection = [[FPersistentConnection alloc] initWithRepoInfo:self.repoInfo dispatchQueue:[FIRDatabaseQuery sharedQueue] config:self.config];
- // Needs to be called before authentication manager is instantiated
- self.eventRaiser = [[FEventRaiser alloc] initWithQueue:self.config.callbackQueue];
- dispatch_async([FIRDatabaseQuery sharedQueue], ^{
- [self deferredInit];
- });
- }
- return self;
- }
- - (void)deferredInit {
- // TODO: cleanup on dealloc
- __weak FRepo *weakSelf = self;
- [self.config.authTokenProvider listenForTokenChanges:^(NSString *token) {
- [weakSelf.connection refreshAuthToken:token];
- }];
- // Open connection now so that by the time we are connected the deferred init has run
- // This relies on the fact that all callbacks run on repos queue
- self.connection.delegate = self;
- [self.connection open];
- self.dataUpdateCount = 0;
- self.rangeMergeUpdateCount = 0;
- self.interceptServerDataCallback = nil;
- if (self.config.persistenceEnabled) {
- NSString* repoHashString = [NSString stringWithFormat:@"%@_%@", self.repoInfo.host, self.repoInfo.namespace];
- NSString* persistencePrefix = [NSString stringWithFormat:@"%@/%@", self.config.sessionIdentifier, repoHashString];
- id<FCachePolicy> cachePolicy = [[FLRUCachePolicy alloc] initWithMaxSize:self.config.persistenceCacheSizeBytes];
- id<FStorageEngine> engine;
- if (self.config.forceStorageEngine != nil) {
- engine = self.config.forceStorageEngine;
- } else {
- FLevelDBStorageEngine *levelDBEngine = [[FLevelDBStorageEngine alloc] initWithPath:persistencePrefix];
- // We need the repo info to run the legacy migration. Future migrations will be managed by the database itself
- // Remove this once we are confident that no-one is using legacy migration anymore...
- [levelDBEngine runLegacyMigration:self.repoInfo];
- engine = levelDBEngine;
- }
- self.persistenceManager = [[FPersistenceManager alloc] initWithStorageEngine:engine cachePolicy:cachePolicy];
- } else {
- self.persistenceManager = nil;
- }
- [self initTransactions];
- // A list of data pieces and paths to be set when this client disconnects
- self.onDisconnect = [[FSparseSnapshotTree alloc] init];
- self.infoData = [[FSnapshotHolder alloc] init];
- FListenProvider *infoListenProvider = [[FListenProvider alloc] init];
- infoListenProvider.startListening = ^(FQuerySpec *query,
- NSNumber *tagId,
- id<FSyncTreeHash> hash,
- fbt_nsarray_nsstring onComplete) {
- NSArray *infoEvents = @[];
- FRepo *strongSelf = weakSelf;
- id<FNode> node = [strongSelf.infoData getNode:query.path];
- // This is possibly a hack, but we have different semantics for .info endpoints. We don't raise null events
- // on initial data...
- if (![node isEmpty]) {
- infoEvents = [strongSelf.infoSyncTree applyServerOverwriteAtPath:query.path newData:node];
- [strongSelf.eventRaiser raiseCallback:^{
- onComplete(kFWPResponseForActionStatusOk);
- }];
- }
- return infoEvents;
- };
- infoListenProvider.stopListening = ^(FQuerySpec *query, NSNumber *tagId) {};
- self.infoSyncTree = [[FSyncTree alloc] initWithListenProvider:infoListenProvider];
- FListenProvider *serverListenProvider = [[FListenProvider alloc] init];
- serverListenProvider.startListening = ^(FQuerySpec *query,
- NSNumber *tagId,
- id<FSyncTreeHash> hash,
- fbt_nsarray_nsstring onComplete) {
- [weakSelf.connection listen:query tagId:tagId hash:hash onComplete:^(NSString *status) {
- NSArray *events = onComplete(status);
- [weakSelf.eventRaiser raiseEvents:events];
- }];
- // No synchronous events for network-backed sync trees
- return @[];
- };
- serverListenProvider.stopListening = ^(FQuerySpec *query, NSNumber *tag) {
- [weakSelf.connection unlisten:query tagId:tag];
- };
- self.serverSyncTree = [[FSyncTree alloc] initWithPersistenceManager:self.persistenceManager
- listenProvider:serverListenProvider];
- [self restoreWrites];
- [self updateInfo:kDotInfoConnected withValue:@NO];
- [self setupNotifications];
- }
- - (void) restoreWrites {
- NSArray *writes = self.persistenceManager.userWrites;
- NSDictionary *serverValues = [FServerValues generateServerValues:self.serverClock];
- __block NSInteger lastWriteId = NSIntegerMin;
- [writes enumerateObjectsUsingBlock:^(FWriteRecord *write, NSUInteger idx, BOOL *stop) {
- NSInteger writeId = write.writeId;
- fbt_void_nsstring_nsstring callback = ^(NSString *status, NSString *errorReason) {
- [self warnIfWriteFailedAtPath:write.path status:status message:@"Persisted write"];
- [self ackWrite:writeId rerunTransactionsAtPath:write.path status:status];
- };
- if (lastWriteId >= writeId) {
- [NSException raise:NSInternalInconsistencyException format:@"Restored writes were not in order!"];
- }
- lastWriteId = writeId;
- self.writeIdCounter = writeId + 1;
- if ([write isOverwrite]) {
- FFLog(@"I-RDB038001", @"Restoring overwrite with id %ld", (long)write.writeId);
- [self.connection putData:[write.overwrite valForExport:YES]
- forPath:[write.path toString]
- withHash:nil
- withCallback:callback];
- id<FNode> resolved = [FServerValues resolveDeferredValueSnapshot:write.overwrite withServerValues:serverValues];
- [self.serverSyncTree applyUserOverwriteAtPath:write.path newData:resolved writeId:writeId isVisible:YES];
- } else {
- FFLog(@"I-RDB038002", @"Restoring merge with id %ld", (long)write.writeId);
- [self.connection mergeData:[write.merge valForExport:YES]
- forPath:[write.path toString]
- withCallback:callback];
- FCompoundWrite *resolved = [FServerValues resolveDeferredValueCompoundWrite:write.merge withServerValues:serverValues];
- [self.serverSyncTree applyUserMergeAtPath:write.path changedChildren:resolved writeId:writeId];
- }
- }];
- }
- - (NSString*)name {
- return self.repoInfo.namespace;
- }
- - (NSString *) description {
- return [self.repoInfo description];
- }
- - (void) interrupt {
- [self.connection interruptForReason:kFInterruptReasonRepoInterrupt];
- }
- - (void) resume {
- [self.connection resumeForReason:kFInterruptReasonRepoInterrupt];
- }
- // NOTE: Typically if you're calling this, you should be in an @autoreleasepool block to make sure that ARC kicks
- // in and cleans up things no longer referenced (i.e. pendingPutsDB).
- - (void) dispose {
- [self.connection interruptForReason:kFInterruptReasonRepoInterrupt];
- // We need to nil out any references to LevelDB, to make sure the
- // LevelDB exclusive locks are released.
- [self.persistenceManager close];
- }
- - (NSInteger) nextWriteId {
- return self->_writeIdCounter++;
- }
- - (NSTimeInterval) serverTime {
- return [self.serverClock currentTime];
- }
- - (void) set:(FPath *)path withNode:(id<FNode>)node withCallback:(fbt_void_nserror_ref)onComplete {
- id value = [node valForExport:YES];
- FFLog(@"I-RDB038003", @"Setting: %@ with %@ pri: %@", [path toString], [value description], [[node getPriority] val]);
- // TODO: Optimize this behavior to either (a) store flag to skip resolving where possible and / or
- // (b) store unresolved paths on JSON parse
- NSDictionary* serverValues = [FServerValues generateServerValues:self.serverClock];
- id<FNode> newNode = [FServerValues resolveDeferredValueSnapshot:node withServerValues:serverValues];
- NSInteger writeId = [self nextWriteId];
- [self.persistenceManager saveUserOverwrite:node atPath:path writeId:writeId];
- NSArray *events = [self.serverSyncTree applyUserOverwriteAtPath:path newData:newNode writeId:writeId isVisible:YES];
- [self.eventRaiser raiseEvents:events];
- [self.connection putData:value forPath:[path toString] withHash:nil withCallback:^(NSString *status, NSString *errorReason) {
- [self warnIfWriteFailedAtPath:path status:status message:@"setValue: or removeValue:"];
- [self ackWrite:writeId rerunTransactionsAtPath:path status:status];
- [self callOnComplete:onComplete withStatus:status errorReason:errorReason andPath:path];
- }];
- FPath* affectedPath = [self abortTransactionsAtPath:path error:kFTransactionSet];
- [self rerunTransactionsForPath:affectedPath];
- }
- - (void) update:(FPath *)path withNodes:(FCompoundWrite *)nodes withCallback:(fbt_void_nserror_ref)callback {
- NSDictionary *values = [nodes valForExport:YES];
- FFLog(@"I-RDB038004", @"Updating: %@ with %@", [path toString], [values description]);
- NSDictionary* serverValues = [FServerValues generateServerValues:self.serverClock];
- FCompoundWrite *resolved = [FServerValues resolveDeferredValueCompoundWrite:nodes withServerValues:serverValues];
- if (!resolved.isEmpty) {
- NSInteger writeId = [self nextWriteId];
- [self.persistenceManager saveUserMerge:nodes atPath:path writeId:writeId];
- NSArray *events = [self.serverSyncTree applyUserMergeAtPath:path changedChildren:resolved writeId:writeId];
- [self.eventRaiser raiseEvents:events];
- [self.connection mergeData:values forPath:[path description] withCallback:^(NSString *status, NSString *errorReason) {
- [self warnIfWriteFailedAtPath:path status:status message:@"updateChildValues:"];
- [self ackWrite:writeId rerunTransactionsAtPath:path status:status];
- [self callOnComplete:callback withStatus:status errorReason:errorReason andPath:path];
- }];
- [nodes enumerateWrites:^(FPath *childPath, id<FNode> node, BOOL *stop) {
- FPath* pathFromRoot = [path child:childPath];
- FFLog(@"I-RDB038005", @"Cancelling transactions at path: %@", pathFromRoot);
- FPath *affectedPath = [self abortTransactionsAtPath:pathFromRoot error:kFTransactionSet];
- [self rerunTransactionsForPath:affectedPath];
- }];
- } else {
- FFLog(@"I-RDB038006", @"update called with empty data. Doing nothing");
- // Do nothing, just call the callback
- [self callOnComplete:callback withStatus:@"ok" errorReason:nil andPath:path];
- }
- }
- - (void) onDisconnectCancel:(FPath *)path withCallback:(fbt_void_nserror_ref)callback {
- [self.connection onDisconnectCancelPath:path withCallback:^(NSString *status, NSString *errorReason) {
- BOOL success = [status isEqualToString:kFWPResponseForActionStatusOk];
- if (success) {
- [self.onDisconnect forgetPath:path];
- } else {
- FFLog(@"I-RDB038007", @"cancelDisconnectOperations: at %@ failed: %@", path, status);
- }
- [self callOnComplete:callback withStatus:status errorReason:errorReason andPath:path];
- }];
- }
- - (void) onDisconnectSet:(FPath *)path withNode:(id<FNode>)node withCallback:(fbt_void_nserror_ref)callback {
- [self.connection onDisconnectPutData:[node valForExport:YES] forPath:path withCallback:^(NSString *status, NSString *errorReason) {
- BOOL success = [status isEqualToString:kFWPResponseForActionStatusOk];
- if (success) {
- [self.onDisconnect rememberData:node onPath:path];
- } else {
- FFWarn(@"I-RDB038008", @"onDisconnectSetValue: or onDisconnectRemoveValue: at %@ failed: %@", path, status);
- }
- [self callOnComplete:callback withStatus:status errorReason:errorReason andPath:path];
- }];
- }
- - (void) onDisconnectUpdate:(FPath *)path withNodes:(FCompoundWrite *)nodes withCallback:(fbt_void_nserror_ref)callback {
- if (!nodes.isEmpty) {
- NSDictionary *values = [nodes valForExport:YES];
- [self.connection onDisconnectMergeData:values forPath:path withCallback:^(NSString *status, NSString *errorReason) {
- BOOL success = [status isEqualToString:kFWPResponseForActionStatusOk];
- if (success) {
- [nodes enumerateWrites:^(FPath *relativePath, id<FNode> nodeUnresolved, BOOL *stop) {
- FPath* childPath = [path child:relativePath];
- [self.onDisconnect rememberData:nodeUnresolved onPath:childPath];
- }];
- } else {
- FFWarn(@"I-RDB038009", @"onDisconnectUpdateChildValues: at %@ failed %@", path, status);
- }
- [self callOnComplete:callback withStatus:status errorReason:errorReason andPath:path];
- }];
- } else {
- // Do nothing, just call the callback
- [self callOnComplete:callback withStatus:@"ok" errorReason:nil andPath:path];
- }
- }
- - (void) purgeOutstandingWrites {
- FFLog(@"I-RDB038010", @"Purging outstanding writes");
- NSArray *events = [self.serverSyncTree removeAllWrites];
- [self.eventRaiser raiseEvents:events];
- // Abort any transactions
- [self abortTransactionsAtPath:[FPath empty] error:kFErrorWriteCanceled];
- // Remove outstanding writes from connection
- [self.connection purgeOutstandingWrites];
- }
- - (void) addEventRegistration:(id <FEventRegistration>)eventRegistration forQuery:(FQuerySpec *)query {
- NSArray *events = nil;
- if ([[query.path getFront] isEqualToString:kDotInfoPrefix]) {
- events = [self.infoSyncTree addEventRegistration:eventRegistration forQuery:query];
- } else {
- events = [self.serverSyncTree addEventRegistration:eventRegistration forQuery:query];
- }
- [self.eventRaiser raiseEvents:events];
- }
- - (void) removeEventRegistration:(id<FEventRegistration>)eventRegistration forQuery:(FQuerySpec *)query {
- // These are guaranteed not to raise events, since we're not passing in a cancelError. However we can future-proof
- // a little bit by handling the return values anyways.
- FFLog(@"I-RDB038011", @"Removing event registration with hande: %lu", (unsigned long)eventRegistration.handle);
- NSArray *events = nil;
- if ([[query.path getFront] isEqualToString:kDotInfoPrefix]) {
- events = [self.infoSyncTree removeEventRegistration:eventRegistration forQuery:query cancelError:nil];
- } else {
- events = [self.serverSyncTree removeEventRegistration:eventRegistration forQuery:query cancelError:nil];
- }
- [self.eventRaiser raiseEvents:events];
- }
- - (void) keepQuery:(FQuerySpec *)query synced:(BOOL)synced {
- NSAssert(![[query.path getFront] isEqualToString:kDotInfoPrefix], @"Can't keep .info tree synced!");
- [self.serverSyncTree keepQuery:query synced:synced];
- }
- - (void) updateInfo:(NSString *) pathString withValue:(id)value {
- // hack to make serverTimeOffset available in a threadsafe way. Property is marked as atomic
- if ([pathString isEqualToString:kDotInfoServerTimeOffset]) {
- NSTimeInterval offset = [(NSNumber *)value doubleValue]/1000.0;
- self.serverClock = [[FOffsetClock alloc] initWithClock:[FSystemClock clock] offset:offset];
- }
- FPath* path = [[FPath alloc] initWith:[NSString stringWithFormat:@"%@/%@", kDotInfoPrefix, pathString]];
- id<FNode> newNode = [FSnapshotUtilities nodeFrom:value];
- [self.infoData updateSnapshot:path withNewSnapshot:newNode];
- NSArray *events = [self.infoSyncTree applyServerOverwriteAtPath:path newData:newNode];
- [self.eventRaiser raiseEvents:events];
- }
- - (void) callOnComplete:(fbt_void_nserror_ref)onComplete withStatus:(NSString *)status errorReason:(NSString *)errorReason andPath:(FPath *)path {
- if (onComplete) {
- FIRDatabaseReference * ref = [[FIRDatabaseReference alloc] initWithRepo:self path:path];
- BOOL statusOk = [status isEqualToString:kFWPResponseForActionStatusOk];
- NSError* err = nil;
- if (!statusOk) {
- err = [FUtilities errorForStatus:status andReason:errorReason];
- }
- [self.eventRaiser raiseCallback:^{
- onComplete(err, ref);
- }];
- }
- }
- - (void)ackWrite:(NSInteger)writeId rerunTransactionsAtPath:(FPath *)path status:(NSString *)status {
- if ([status isEqualToString:kFErrorWriteCanceled]) {
- // This write was already removed, we just need to ignore it...
- } else {
- BOOL success = [status isEqualToString:kFWPResponseForActionStatusOk];
- NSArray *clearEvents = [self.serverSyncTree ackUserWriteWithWriteId:writeId revert:!success persist:YES clock:self.serverClock];
- if ([clearEvents count] > 0) {
- [self rerunTransactionsForPath:path];
- }
- [self.eventRaiser raiseEvents:clearEvents];
- }
- }
- - (void) warnIfWriteFailedAtPath:(FPath *)path status:(NSString *)status message:(NSString *)message {
- if (!([status isEqualToString:kFWPResponseForActionStatusOk] || [status isEqualToString:kFErrorWriteCanceled])) {
- FFWarn(@"I-RDB038012", @"%@ at %@ failed: %@", message, path, status);
- }
- }
- #pragma mark -
- #pragma mark FPersistentConnectionDelegate methods
- - (void) onDataUpdate:(FPersistentConnection *)fpconnection forPath:(NSString *)pathString message:(id)data isMerge:(BOOL)isMerge tagId:(NSNumber *)tagId {
- FFLog(@"I-RDB038013", @"onDataUpdateForPath: %@ withMessage: %@", pathString, data);
- // For testing.
- self.dataUpdateCount++;
- FPath* path = [[FPath alloc] initWith:pathString];
- data = self.interceptServerDataCallback ? self.interceptServerDataCallback(pathString, data) : data;
- NSArray *events = nil;
- if (tagId != nil) {
- if (isMerge) {
- NSDictionary *message = data;
- FCompoundWrite *taggedChildren = [FCompoundWrite compoundWriteWithValueDictionary:message];
- events = [self.serverSyncTree applyTaggedQueryMergeAtPath:path changedChildren:taggedChildren tagId:tagId];
- } else {
- id<FNode> taggedSnap = [FSnapshotUtilities nodeFrom:data];
- events = [self.serverSyncTree applyTaggedQueryOverwriteAtPath:path newData:taggedSnap tagId:tagId];
- }
- } else if (isMerge) {
- NSDictionary *message = data;
- FCompoundWrite *changedChildren = [FCompoundWrite compoundWriteWithValueDictionary:message];
- events = [self.serverSyncTree applyServerMergeAtPath:path changedChildren:changedChildren];
- } else {
- id<FNode> snap = [FSnapshotUtilities nodeFrom:data];
- events = [self.serverSyncTree applyServerOverwriteAtPath:path newData:snap];
- }
- if ([events count] > 0) {
- // Since we have a listener outstanding for each transaction, receiving any events
- // is a proxy for some change having occurred.
- [self rerunTransactionsForPath:path];
- }
- [self.eventRaiser raiseEvents:events];
- }
- - (void)onRangeMerge:(NSArray *)ranges forPath:(NSString *)pathString tagId:(NSNumber *)tag {
- FFLog(@"I-RDB038014", @"onRangeMerge: %@ => %@", pathString, ranges);
- // For testing
- self.rangeMergeUpdateCount++;
- FPath* path = [[FPath alloc] initWith:pathString];
- NSArray *events;
- if (tag != nil) {
- events = [self.serverSyncTree applyTaggedServerRangeMergeAtPath:path updates:ranges tagId:tag];
- } else {
- events = [self.serverSyncTree applyServerRangeMergeAtPath:path updates:ranges];
- }
- if (events.count > 0) {
- // Since we have a listener outstanding for each transaction, receiving any events
- // is a proxy for some change having occurred.
- [self rerunTransactionsForPath:path];
- }
- [self.eventRaiser raiseEvents:events];
- }
- - (void)onConnect:(FPersistentConnection *)fpconnection {
- [self updateInfo:kDotInfoConnected withValue:@YES];
- }
- - (void)onDisconnect:(FPersistentConnection *)fpconnection {
- [self updateInfo:kDotInfoConnected withValue:@NO];
- [self runOnDisconnectEvents];
- }
- - (void)onServerInfoUpdate:(FPersistentConnection *)fpconnection updates:(NSDictionary *)updates {
- for (NSString* key in updates) {
- id val = [updates objectForKey:key];
- [self updateInfo:key withValue:val];
- }
- }
- - (void) setupNotifications {
- NSString * const *backgroundConstant = (NSString * const *) dlsym(RTLD_DEFAULT, "UIApplicationDidEnterBackgroundNotification");
- if (backgroundConstant) {
- FFLog(@"I-RDB038015", @"Registering for background notification.");
- [[NSNotificationCenter defaultCenter] addObserver:self
- selector:@selector(didEnterBackground)
- name:*backgroundConstant
- object:nil];
- } else {
- FFLog(@"I-RDB038016", @"Skipped registering for background notification.");
- }
- }
- - (void) didEnterBackground {
- if (!self.config.persistenceEnabled)
- return;
- // Targetted compilation is ONLY for testing. UIKit is weak-linked in actual release build.
- #if TARGET_OS_IOS || TARGET_OS_TV
- // The idea is to wait until any outstanding sets get written to disk. Since the sets might still be in our
- // dispatch queue, we wait for the dispatch queue to catch up and for persistence to catch up.
- // This may be undesirable though. The dispatch queue might just be processing a bunch of incoming data or
- // something. We might want to keep track of whether there are any unpersisted sets or something.
- FFLog(@"I-RDB038017", @"Entering background. Starting background task to finish work.");
- Class uiApplicationClass = NSClassFromString(@"UIApplication");
- assert(uiApplicationClass); // If we are here, we should be on iOS and UIApplication should be available.
- UIApplication *application = [uiApplicationClass sharedApplication];
- __block UIBackgroundTaskIdentifier bgTask = [application beginBackgroundTaskWithExpirationHandler:^{
- [application endBackgroundTask:bgTask];
- }];
- NSDate *start = [NSDate date];
- dispatch_async([FIRDatabaseQuery sharedQueue], ^{
- NSTimeInterval finishTime = [start timeIntervalSinceNow]*-1;
- FFLog(@"I-RDB038018", @"Background task completed. Queue time: %f", finishTime);
- [application endBackgroundTask:bgTask];
- });
- #endif
- }
- #pragma mark -
- #pragma mark Internal methods
- /**
- * Applies all the changes stored up in the onDisconnect tree
- */
- - (void) runOnDisconnectEvents {
- FFLog(@"I-RDB038019", @"Running onDisconnectEvents");
- NSDictionary* serverValues = [FServerValues generateServerValues:self.serverClock];
- FSparseSnapshotTree* resolvedTree = [FServerValues resolveDeferredValueTree:self.onDisconnect withServerValues:serverValues];
- NSMutableArray *events = [[NSMutableArray alloc] init];
- [resolvedTree forEachTreeAtPath:[FPath empty] do:^(FPath *path, id<FNode> node) {
- [events addObjectsFromArray:[self.serverSyncTree applyServerOverwriteAtPath:path newData:node]];
- FPath* affectedPath = [self abortTransactionsAtPath:path error:kFTransactionSet];
- [self rerunTransactionsForPath:affectedPath];
- }];
- self.onDisconnect = [[FSparseSnapshotTree alloc] init];
- [self.eventRaiser raiseEvents:events];
- }
- - (NSDictionary *) dumpListens {
- return [self.connection dumpListens];
- }
- #pragma mark -
- #pragma mark Transactions
- /**
- * Setup the transaction data structures
- */
- - (void) initTransactions {
- self.transactionQueueTree = [[FTree alloc] init];
- self.hijackHash = NO;
- self.loggedTransactionPersistenceWarning = NO;
- }
- /**
- * Creates a new transaction, add its to the transactions we're tracking, and sends it to the server if possible
- */
- - (void) startTransactionOnPath:(FPath *)path update:(fbt_transactionresult_mutabledata)update onComplete:(fbt_void_nserror_bool_datasnapshot)onComplete withLocalEvents:(BOOL)applyLocally {
- if (self.config.persistenceEnabled && !self.loggedTransactionPersistenceWarning) {
- self.loggedTransactionPersistenceWarning = YES;
- FFInfo(@"I-RDB038020", @"runTransactionBlock: usage detected while persistence is enabled. Please be aware that transactions "
- @"*will not* be persisted across app restarts. "
- @"See https://www.firebase.com/docs/ios/guide/offline-capabilities.html#section-handling-transactions-offline for more details.");
- }
- FIRDatabaseReference * watchRef = [[FIRDatabaseReference alloc] initWithRepo:self path:path];
- // make sure we're listening on this node
- // Note: we can't do this asynchronously. To preserve event ordering, it has to be done in this block.
- // This is ok, this block is guaranteed to be our own event loop
- NSUInteger handle = [[FUtilities LUIDGenerator] integerValue];
- fbt_void_datasnapshot cb = ^(FIRDataSnapshot *snapshot) {};
- FValueEventRegistration *registration = [[FValueEventRegistration alloc] initWithRepo:self
- handle:handle
- callback:cb
- cancelCallback:nil];
- [watchRef.repo addEventRegistration:registration forQuery:watchRef.querySpec];
- fbt_void_void unwatcher = ^{ [watchRef removeObserverWithHandle:handle]; };
- // Save all the data that represents this transaction
- FTupleTransaction* transaction = [[FTupleTransaction alloc] init];
- transaction.path = path;
- transaction.update = update;
- transaction.onComplete = onComplete;
- transaction.status = FTransactionInitializing;
- transaction.order = [FUtilities LUIDGenerator];
- transaction.applyLocally = applyLocally;
- transaction.retryCount = 0;
- transaction.unwatcher = unwatcher;
- transaction.currentWriteId = nil;
- transaction.currentInputSnapshot = nil;
- transaction.currentOutputSnapshotRaw = nil;
- transaction.currentOutputSnapshotResolved = nil;
- // Run transaction initially
- id<FNode> currentState = [self latestStateAtPath:path excludeWriteIds:nil];
- transaction.currentInputSnapshot = currentState;
- FIRMutableData * mutableCurrent = [[FIRMutableData alloc] initWithNode:currentState];
- FIRTransactionResult * result = transaction.update(mutableCurrent);
- if (!result.isSuccess) {
- // Abort the transaction
- transaction.unwatcher();
- transaction.currentOutputSnapshotRaw = nil;
- transaction.currentOutputSnapshotResolved = nil;
- if (transaction.onComplete) {
- FIRDatabaseReference *ref = [[FIRDatabaseReference alloc] initWithRepo:self path:transaction.path];
- FIndexedNode *indexedNode = [FIndexedNode indexedNodeWithNode:transaction.currentInputSnapshot];
- FIRDataSnapshot *snap = [[FIRDataSnapshot alloc] initWithRef:ref indexedNode:indexedNode];
- [self.eventRaiser raiseCallback:^{
- transaction.onComplete(nil, NO, snap);
- }];
- }
- } else {
- // Note: different from js. We don't need to validate, FIRMutableData does validation.
- // We also don't have to worry about priorities. Just mark as run and add to queue.
- transaction.status = FTransactionRun;
- FTree* queueNode = [self.transactionQueueTree subTree:transaction.path];
- NSMutableArray* nodeQueue = [queueNode getValue];
- if (nodeQueue == nil) {
- nodeQueue = [[NSMutableArray alloc] init];
- }
- [nodeQueue addObject:transaction];
- [queueNode setValue:nodeQueue];
- // Update visibleData and raise events
- // Note: We intentionally raise events after updating all of our transaction state, since the user could
- // start new transactions from the event callbacks
- NSDictionary* serverValues = [FServerValues generateServerValues:self.serverClock];
- id<FNode> newValUnresolved = [result.update nodeValue];
- id<FNode> newVal = [FServerValues resolveDeferredValueSnapshot:newValUnresolved withServerValues:serverValues];
- transaction.currentOutputSnapshotRaw = newValUnresolved;
- transaction.currentOutputSnapshotResolved = newVal;
- transaction.currentWriteId = [NSNumber numberWithInteger:[self nextWriteId]];
- NSArray *events = [self.serverSyncTree applyUserOverwriteAtPath:path newData:newVal
- writeId:[transaction.currentWriteId integerValue]
- isVisible:transaction.applyLocally];
- [self.eventRaiser raiseEvents:events];
- [self sendAllReadyTransactions];
- }
- }
- /**
- * @param writeIdsToExclude A specific set to exclude
- */
- - (id<FNode>) latestStateAtPath:(FPath *)path excludeWriteIds:(NSArray *)writeIdsToExclude {
- id<FNode> latestState = [self.serverSyncTree calcCompleteEventCacheAtPath:path excludeWriteIds:writeIdsToExclude];
- return latestState ? latestState : [FEmptyNode emptyNode];
- }
- /**
- * Sends any already-run transactions that aren't waiting for outstanding transactions to complete.
- *
- * Externally, call the version with no arguments.
- * Internally, calls itself recursively with a particular transactionQueueTree node to recurse through the tree
- */
- - (void) sendAllReadyTransactions {
- FTree* node = self.transactionQueueTree;
- [self pruneCompletedTransactionsBelowNode:node];
- [self sendReadyTransactionsForTree:node];
- }
- - (void) sendReadyTransactionsForTree:(FTree *)node {
- NSMutableArray* queue = [node getValue];
- if (queue != nil) {
- queue = [self buildTransactionQueueAtNode:node];
- NSAssert([queue count] > 0, @"Sending zero length transaction queue");
- NSUInteger notRunIndex = [queue indexOfObjectPassingTest:^BOOL(id obj, NSUInteger idx, BOOL *stop) {
- return ((FTupleTransaction*)obj).status != FTransactionRun;
- }];
- // If they're all run (and not sent), we can send them. Else, we must wait.
- if (notRunIndex == NSNotFound) {
- [self sendTransactionQueue:queue atPath:node.path];
- }
- } else if ([node hasChildren]) {
- [node forEachChild:^(FTree *child) {
- [self sendReadyTransactionsForTree:child];
- }];
- }
- }
- /**
- * Given a list of run transactions, send them to the server and then handle the result (success or failure).
- */
- - (void) sendTransactionQueue:(NSMutableArray *)queue atPath:(FPath *)path {
- // Mark transactions as sent and bump the retry count
- NSMutableArray *writeIdsToExclude = [[NSMutableArray alloc] init];
- for (FTupleTransaction *transaction in queue) {
- [writeIdsToExclude addObject:transaction.currentWriteId];
- }
- id<FNode> latestState = [self latestStateAtPath:path excludeWriteIds:writeIdsToExclude];
- id<FNode> snapToSend = latestState;
- NSString *latestHash = [latestState dataHash];
- for (FTupleTransaction* transaction in queue) {
- NSAssert(transaction.status == FTransactionRun, @"[FRepo sendTransactionQueue:] items in queue should all be run.");
- FFLog(@"I-RDB038021", @"Transaction at %@ set to SENT", transaction.path);
- transaction.status = FTransactionSent;
- transaction.retryCount++;
- FPath *relativePath = [FPath relativePathFrom:path to:transaction.path];
- // If we've gotten to this point, the output snapshot must be defined.
- snapToSend = [snapToSend updateChild:relativePath withNewChild:transaction.currentOutputSnapshotRaw];
- }
- id dataToSend = [snapToSend valForExport:YES];
- NSString *pathToSend = [path description];
- latestHash = self.hijackHash ? @"badhash" : latestHash;
- // Send the put
- [self.connection putData:dataToSend forPath:pathToSend withHash:latestHash withCallback:^(NSString *status, NSString *errorReason) {
- FFLog(@"I-RDB038022", @"Transaction put response: %@ : %@", pathToSend, status);
- NSMutableArray *events = [[NSMutableArray alloc] init];
- if ([status isEqualToString:kFWPResponseForActionStatusOk]) {
- // Queue up the callbacks and fire them after cleaning up all of our transaction state, since
- // the callback could trigger more transactions or sets.
- NSMutableArray *callbacks = [[NSMutableArray alloc] init];
- for (FTupleTransaction *transaction in queue) {
- transaction.status = FTransactionCompleted;
- [events addObjectsFromArray:[self.serverSyncTree ackUserWriteWithWriteId:[transaction.currentWriteId integerValue]
- revert:NO
- persist:NO
- clock:self.serverClock]];
- if (transaction.onComplete) {
- // We never unset the output snapshot, and given that this transaction is complete, it should be set
- id <FNode> node = transaction.currentOutputSnapshotResolved;
- FIndexedNode *indexedNode = [FIndexedNode indexedNodeWithNode:node];
- FIRDatabaseReference *ref = [[FIRDatabaseReference alloc] initWithRepo:self path:transaction.path];
- FIRDataSnapshot *snapshot = [[FIRDataSnapshot alloc] initWithRef:ref indexedNode:indexedNode];
- fbt_void_void cb = ^{
- transaction.onComplete(nil, YES, snapshot);
- };
- [callbacks addObject:[cb copy]];
- }
- transaction.unwatcher();
- }
- // Now remove the completed transactions.
- [self pruneCompletedTransactionsBelowNode:[self.transactionQueueTree subTree:path]];
- // There may be pending transactions that we can now send.
- [self sendAllReadyTransactions];
- // Finally, trigger onComplete callbacks
- [self.eventRaiser raiseCallbacks:callbacks];
- } else {
- // transactions are no longer sent. Update their status appropriately.
- if ([status isEqualToString:kFWPResponseForActionStatusDataStale]) {
- for (FTupleTransaction *transaction in queue) {
- if (transaction.status == FTransactionSentNeedsAbort) {
- transaction.status = FTransactionNeedsAbort;
- } else {
- transaction.status = FTransactionRun;
- }
- }
- } else {
- FFWarn(@"I-RDB038023", @"runTransactionBlock: at %@ failed: %@", path, status);
- for (FTupleTransaction *transaction in queue) {
- transaction.status = FTransactionNeedsAbort;
- [transaction setAbortStatus:status reason:errorReason];
- }
- }
- }
- [self rerunTransactionsForPath:path];
- [self.eventRaiser raiseEvents:events];
- }];
- }
- /**
- * Finds all transactions dependent on the data at changed Path and reruns them.
- *
- * Should be called any time cached data changes.
- *
- * Return the highest path that was affected by rerunning transactions. This is the path at which events need to
- * be raised for.
- */
- - (FPath *) rerunTransactionsForPath:(FPath *)changedPath {
- // For the common case that there are no transactions going on, skip all this!
- if ([self.transactionQueueTree isEmpty]) {
- return changedPath;
- } else {
- FTree* rootMostTransactionNode = [self getAncestorTransactionNodeForPath:changedPath];
- FPath* path = rootMostTransactionNode.path;
- NSArray* queue = [self buildTransactionQueueAtNode:rootMostTransactionNode];
- [self rerunTransactionQueue:queue atPath:path];
- return path;
- }
- }
- /**
- * Does all the work of rerunning transactions (as well as cleans up aborted transactions and whatnot).
- */
- - (void) rerunTransactionQueue:(NSArray *)queue atPath:(FPath *)path {
- if (queue.count == 0) {
- return; // nothing to do
- }
- // Queue up the callbacks and fire them after cleaning up all of our transaction state, since
- // the callback could trigger more transactions or sets.
- NSMutableArray *events = [[NSMutableArray alloc] init];
- NSMutableArray *callbacks = [[NSMutableArray alloc] init];
- // Ignore, by default, all of the sets in this queue, since we're re-running all of them. However, we want to include
- // the results of new sets triggered as part of this re-run, so we don't want to ignore a range, just these specific
- // sets.
- NSMutableArray *writeIdsToExclude = [[NSMutableArray alloc] init];
- for (FTupleTransaction *transaction in queue) {
- [writeIdsToExclude addObject:transaction.currentWriteId];
- }
- for (FTupleTransaction* transaction in queue) {
- FPath* relativePath __unused = [FPath relativePathFrom:path to:transaction.path];
- BOOL abortTransaction = NO;
- NSAssert(relativePath != nil, @"[FRepo rerunTransactionsQueue:] relativePath should not be null.");
- if (transaction.status == FTransactionNeedsAbort) {
- abortTransaction = YES;
- if (![transaction.abortStatus isEqualToString:kFErrorWriteCanceled]) {
- NSArray *ackEvents = [self.serverSyncTree ackUserWriteWithWriteId:[transaction.currentWriteId integerValue]
- revert:YES
- persist:NO
- clock:self.serverClock];
- [events addObjectsFromArray:ackEvents];
- }
- } else if (transaction.status == FTransactionRun) {
- if (transaction.retryCount >= kFTransactionMaxRetries) {
- abortTransaction = YES;
- [transaction setAbortStatus:kFTransactionTooManyRetries reason:nil];
- [events addObjectsFromArray:[self.serverSyncTree ackUserWriteWithWriteId:[transaction.currentWriteId integerValue]
- revert:YES
- persist:NO
- clock:self.serverClock]];
- } else {
- // This code reruns a transaction
- id<FNode> currentNode = [self latestStateAtPath:transaction.path excludeWriteIds:writeIdsToExclude];
- transaction.currentInputSnapshot = currentNode;
- FIRMutableData * mutableCurrent = [[FIRMutableData alloc] initWithNode:currentNode];
- FIRTransactionResult * result = transaction.update(mutableCurrent);
- if (result.isSuccess) {
- NSNumber *oldWriteId = transaction.currentWriteId;
- NSDictionary* serverValues = [FServerValues generateServerValues:self.serverClock];
- id<FNode> newVal = [result.update nodeValue];
- id<FNode> newValResolved = [FServerValues resolveDeferredValueSnapshot:newVal withServerValues:serverValues];
- transaction.currentOutputSnapshotRaw = newVal;
- transaction.currentOutputSnapshotResolved = newValResolved;
- transaction.currentWriteId = [NSNumber numberWithInteger:[self nextWriteId]];
- // Mutates writeIdsToExclude in place
- [writeIdsToExclude removeObject:oldWriteId];
- [events addObjectsFromArray:[self.serverSyncTree applyUserOverwriteAtPath:transaction.path
- newData:transaction.currentOutputSnapshotResolved
- writeId:[transaction.currentWriteId integerValue]
- isVisible:transaction.applyLocally]];
- [events addObjectsFromArray:[self.serverSyncTree ackUserWriteWithWriteId:[oldWriteId integerValue]
- revert:YES
- persist:NO
- clock:self.serverClock]];
- } else {
- abortTransaction = YES;
- // The user aborted the transaction. JS treats ths as a "nodata" abort, but it's not an error, so we don't send them an error.
- [transaction setAbortStatus:nil reason:nil];
- [events addObjectsFromArray:[self.serverSyncTree ackUserWriteWithWriteId:[transaction.currentWriteId integerValue]
- revert:YES
- persist:NO
- clock:self.serverClock]];
- }
- }
- }
- [self.eventRaiser raiseEvents:events];
- events = nil;
- if (abortTransaction) {
- // Abort
- transaction.status = FTransactionCompleted;
- transaction.unwatcher();
- if (transaction.onComplete) {
- FIRDatabaseReference * ref = [[FIRDatabaseReference alloc] initWithRepo:self path:transaction.path];
- FIndexedNode *lastInput = [FIndexedNode indexedNodeWithNode:transaction.currentInputSnapshot];
- FIRDataSnapshot * snap = [[FIRDataSnapshot alloc] initWithRef:ref indexedNode:lastInput];
- fbt_void_void cb = ^{
- // Unlike JS, no need to check for "nodata" because ObjC has abortError = nil
- transaction.onComplete(transaction.abortError, NO, snap);
- };
- [callbacks addObject:[cb copy]];
- }
- }
- }
- // Note: unlike current js client, we don't need to preserve priority. Users can set priority via FIRMutableData
- // Clean up completed transactions.
- [self pruneCompletedTransactionsBelowNode:self.transactionQueueTree];
- // Now fire callbacks, now that we're in a good, known state.
- [self.eventRaiser raiseCallbacks:callbacks];
- // Try to send the transaction result to the server
- [self sendAllReadyTransactions];
- }
- - (FTree *) getAncestorTransactionNodeForPath:(FPath *)path {
- FTree* transactionNode = self.transactionQueueTree;
- while (![path isEmpty] && [transactionNode getValue] == nil) {
- NSString* front = [path getFront];
- transactionNode = [transactionNode subTree:[[FPath alloc] initWith:front]];
- path = [path popFront];
- }
- return transactionNode;
- }
- - (NSMutableArray *) buildTransactionQueueAtNode:(FTree *)node {
- NSMutableArray* queue = [[NSMutableArray alloc] init];
- [self aggregateTransactionQueuesForNode:node andQueue:queue];
- [queue sortUsingComparator:^NSComparisonResult(FTupleTransaction* obj1, FTupleTransaction* obj2) {
- return [obj1.order compare:obj2.order];
- }];
- return queue;
- }
- - (void) aggregateTransactionQueuesForNode:(FTree *)node andQueue:(NSMutableArray *)queue {
- NSArray* nodeQueue = [node getValue];
- [queue addObjectsFromArray:nodeQueue];
- [node forEachChild:^(FTree *child) {
- [self aggregateTransactionQueuesForNode:child andQueue:queue];
- }];
- }
- /**
- * Remove COMPLETED transactions at or below this node in the transactionQueueTree
- */
- - (void) pruneCompletedTransactionsBelowNode:(FTree *)node {
- NSMutableArray* queue = [node getValue];
- if (queue != nil) {
- int i = 0;
- // remove all of the completed transactions from the queue
- while (i < queue.count) {
- FTupleTransaction* transaction = [queue objectAtIndex:i];
- if (transaction.status == FTransactionCompleted) {
- [queue removeObjectAtIndex:i];
- } else {
- i++;
- }
- }
- if (queue.count > 0) {
- [node setValue:queue];
- } else {
- [node setValue:nil];
- }
- }
- [node forEachChildMutationSafe:^(FTree *child) {
- [self pruneCompletedTransactionsBelowNode:child];
- }];
- }
- /**
- * Aborts all transactions on ancestors or descendants of the specified path. Called when doing a setValue: or
- * updateChildValues: since we consider them incompatible with transactions
- *
- * @param path path for which we want to abort related transactions.
- */
- - (FPath *) abortTransactionsAtPath:(FPath *)path error:(NSString *)error {
- // For the common case that there are no transactions going on, skip all this!
- if ([self.transactionQueueTree isEmpty]) {
- return path;
- } else {
- FPath* affectedPath = [self getAncestorTransactionNodeForPath:path].path;
- FTree* transactionNode = [self.transactionQueueTree subTree:path];
- [transactionNode forEachAncestor:^BOOL(FTree *ancestor) {
- [self abortTransactionsAtNode:ancestor error:error];
- return NO;
- }];
- [self abortTransactionsAtNode:transactionNode error:error];
- [transactionNode forEachDescendant:^(FTree *child) {
- [self abortTransactionsAtNode:child error:error];
- }];
- return affectedPath;
- }
- }
- /**
- * Abort transactions stored in this transactions queue node.
- *
- * @param node Node to abort transactions for.
- */
- - (void) abortTransactionsAtNode:(FTree *)node error:(NSString *)error {
- NSMutableArray* queue = [node getValue];
- if (queue != nil) {
- // Queue up the callbacks and fire them after cleaning up all of our transaction state, since
- // can be immediately aborted and removed.
- NSMutableArray* callbacks = [[NSMutableArray alloc] init];
- // Go through queue. Any already-sent transactions must be marked for abort, while the unsent ones
- // can be immediately aborted and removed
- NSMutableArray *events = [[NSMutableArray alloc] init];
- int lastSent = -1;
- // Note: all of the sent transactions will be at the front of the queue, so safe to increment lastSent
- for (FTupleTransaction* transaction in queue) {
- if (transaction.status == FTransactionSentNeedsAbort) {
- // No-op. already marked.
- } else if (transaction.status == FTransactionSent) {
- // Mark this transaction for abort when it returns
- lastSent++;
- transaction.status = FTransactionSentNeedsAbort;
- [transaction setAbortStatus:error reason:nil];
- } else {
- // we can abort this immediately
- transaction.unwatcher();
- if ([error isEqualToString:kFTransactionSet]) {
- [events addObjectsFromArray:[self.serverSyncTree ackUserWriteWithWriteId:[transaction.currentWriteId integerValue]
- revert:YES
- persist:NO
- clock:self.serverClock]];
- } else {
- // If it was cancelled it was already removed from the sync tree, no need to ack
- NSAssert([error isEqualToString:kFErrorWriteCanceled], nil);
- }
- if (transaction.onComplete) {
- NSError* abortReason = [FUtilities errorForStatus:error andReason:nil];
- FIRDataSnapshot * snapshot = nil;
- fbt_void_void cb = ^{
- transaction.onComplete(abortReason, NO, snapshot);
- };
- [callbacks addObject:[cb copy]];
- }
- }
- }
- if (lastSent == -1) {
- // We're not waiting for any sent transactions. We can clear the queue.
- [node setValue:nil];
- } else {
- // Remove the transactions we aborted
- NSRange theRange;
- theRange.location = lastSent + 1;
- theRange.length = queue.count - theRange.location;
- [queue removeObjectsInRange:theRange];
- }
- // Now fire the callbacks
- [self.eventRaiser raiseEvents:events];
- [self.eventRaiser raiseCallbacks:callbacks];
- }
- }
- @end
|