819 lines
36 KiB
Objective-C
819 lines
36 KiB
Objective-C
/*
|
|
* 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 <FirebaseCore/FIRLogger.h>
|
|
#import "FSyncTree.h"
|
|
#import "FListenProvider.h"
|
|
#import "FWriteTree.h"
|
|
#import "FNode.h"
|
|
#import "FPath.h"
|
|
#import "FEventRegistration.h"
|
|
#import "FImmutableTree.h"
|
|
#import "FOperation.h"
|
|
#import "FWriteTreeRef.h"
|
|
#import "FOverwrite.h"
|
|
#import "FOperationSource.h"
|
|
#import "FMerge.h"
|
|
#import "FAckUserWrite.h"
|
|
#import "FView.h"
|
|
#import "FSyncPoint.h"
|
|
#import "FEmptyNode.h"
|
|
#import "FQueryParams.h"
|
|
#import "FQuerySpec.h"
|
|
#import "FSnapshotHolder.h"
|
|
#import "FChildrenNode.h"
|
|
#import "FTupleRemovedQueriesEvents.h"
|
|
#import "FAtomicNumber.h"
|
|
#import "FEventRaiser.h"
|
|
#import "FListenComplete.h"
|
|
#import "FSnapshotUtilities.h"
|
|
#import "FCacheNode.h"
|
|
#import "FUtilities.h"
|
|
#import "FCompoundWrite.h"
|
|
#import "FWriteRecord.h"
|
|
#import "FPersistenceManager.h"
|
|
#import "FKeepSyncedEventRegistration.h"
|
|
#import "FServerValues.h"
|
|
#import "FCompoundHash.h"
|
|
#import "FRangeMerge.h"
|
|
|
|
// Size after which we start including the compound hash
|
|
static const NSUInteger kFSizeThresholdForCompoundHash = 1024;
|
|
|
|
@interface FListenContainer : NSObject<FSyncTreeHash>
|
|
|
|
@property (nonatomic, strong) FView *view;
|
|
@property (nonatomic, copy) fbt_nsarray_nsstring onComplete;
|
|
|
|
@end
|
|
|
|
@implementation FListenContainer
|
|
|
|
- (instancetype)initWithView:(FView *)view onComplete:(fbt_nsarray_nsstring)onComplete {
|
|
self = [super init];
|
|
if (self != nil) {
|
|
self->_view = view;
|
|
self->_onComplete = onComplete;
|
|
}
|
|
return self;
|
|
}
|
|
|
|
- (id<FNode>)serverCache {
|
|
return self.view.serverCache;
|
|
}
|
|
|
|
- (FCompoundHash *)compoundHash {
|
|
return [FCompoundHash fromNode:[self serverCache]];
|
|
}
|
|
|
|
- (NSString *)simpleHash {
|
|
return [[self serverCache] dataHash];
|
|
}
|
|
|
|
- (BOOL)includeCompoundHash {
|
|
return [FSnapshotUtilities estimateSerializedNodeSize:[self serverCache]] > kFSizeThresholdForCompoundHash;
|
|
}
|
|
|
|
@end
|
|
|
|
@interface FSyncTree ()
|
|
|
|
/**
|
|
* Tree of SyncPoints. There's a SyncPoint at any location that has 1 or more views.
|
|
*/
|
|
@property (nonatomic, strong) FImmutableTree *syncPointTree;
|
|
|
|
/**
|
|
* A tree of all pending user writes (user-initiated set, transactions, updates, etc)
|
|
*/
|
|
@property (nonatomic, strong) FWriteTree *pendingWriteTree;
|
|
|
|
/**
|
|
* Maps tagId -> FTuplePathQueryParams
|
|
*/
|
|
@property (nonatomic, strong) NSMutableDictionary *tagToQueryMap;
|
|
@property (nonatomic, strong) NSMutableDictionary *queryToTagMap;
|
|
@property (nonatomic, strong) FListenProvider *listenProvider;
|
|
@property (nonatomic, strong) FPersistenceManager *persistenceManager;
|
|
@property (nonatomic, strong) FAtomicNumber *queryTagCounter;
|
|
@property (nonatomic, strong) NSMutableSet *keepSyncedQueries;
|
|
|
|
@end
|
|
|
|
/**
|
|
* SyncTree is the central class for managing event callback registration, data caching, views
|
|
* (query processing), and event generation. There are typically two SyncTree instances for
|
|
* each Repo, one for the normal Firebase data, and one for the .info data.
|
|
*
|
|
* It has a number of responsibilities, including:
|
|
* - Tracking all user event callbacks (registered via addEventRegistration: and removeEventRegistration:).
|
|
* - Applying and caching data changes for user setValue:, runTransactionBlock:, and updateChildValues: calls
|
|
* (applyUserOverwriteAtPath:, applyUserMergeAtPath:).
|
|
* - Applying and caching data changes for server data changes (applyServerOverwriteAtPath:,
|
|
* applyServerMergeAtPath:).
|
|
* - Generating user-facing events for server and user changes (all of the apply* methods
|
|
* return the set of events that need to be raised as a result).
|
|
* - Maintaining the appropriate set of server listens to ensure we are always subscribed
|
|
* to the correct set of paths and queries to satisfy the current set of user event
|
|
* callbacks (listens are started/stopped using the provided listenProvider).
|
|
*
|
|
* NOTE: Although SyncTree tracks event callbacks and calculates events to raise, the actual
|
|
* events are returned to the caller rather than raised synchronously.
|
|
*/
|
|
@implementation FSyncTree
|
|
|
|
- (id) initWithListenProvider:(FListenProvider *)provider {
|
|
return [self initWithPersistenceManager:nil listenProvider:provider];
|
|
}
|
|
|
|
- (id) initWithPersistenceManager:(FPersistenceManager *)persistenceManager listenProvider:(FListenProvider *)provider {
|
|
self = [super init];
|
|
if (self) {
|
|
self.syncPointTree = [FImmutableTree empty];
|
|
self.pendingWriteTree = [[FWriteTree alloc] init];
|
|
self.tagToQueryMap = [[NSMutableDictionary alloc] init];
|
|
self.queryToTagMap = [[NSMutableDictionary alloc] init];
|
|
self.listenProvider = provider;
|
|
self.persistenceManager = persistenceManager;
|
|
self.queryTagCounter = [[FAtomicNumber alloc] init];
|
|
self.keepSyncedQueries = [NSMutableSet set];
|
|
}
|
|
return self;
|
|
}
|
|
|
|
#pragma mark -
|
|
#pragma mark Apply Operations
|
|
|
|
/**
|
|
* Apply data changes for a user-generated setValue: runTransactionBlock: updateChildValues:, etc.
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyUserOverwriteAtPath:(FPath *)path newData:(id <FNode>)newData writeId:(NSInteger)writeId isVisible:(BOOL)visible {
|
|
// Record pending write
|
|
[self.pendingWriteTree addOverwriteAtPath:path newData:newData writeId:writeId isVisible:visible];
|
|
if (!visible) {
|
|
return @[];
|
|
} else {
|
|
FOverwrite *operation = [[FOverwrite alloc] initWithSource:[FOperationSource userInstance] path:path snap:newData];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Apply the data from a user-generated updateChildValues: call
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyUserMergeAtPath:(FPath *)path changedChildren:(FCompoundWrite *)changedChildren writeId:(NSInteger)writeId {
|
|
// Record pending merge
|
|
[self.pendingWriteTree addMergeAtPath:path changedChildren:changedChildren writeId:writeId];
|
|
|
|
FMerge *operation = [[FMerge alloc] initWithSource:[FOperationSource userInstance] path:path children:changedChildren];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
|
|
/**
|
|
* Acknowledge a pending user write that was previously registered with applyUserOverwriteAtPath: or applyUserMergeAtPath:
|
|
* TODO[offline]: Taking a serverClock here is awkward, but server values are awkward. :-(
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) ackUserWriteWithWriteId:(NSInteger)writeId revert:(BOOL)revert persist:(BOOL)persist clock:(id<FClock>)clock {
|
|
FWriteRecord *write = [self.pendingWriteTree writeForId:writeId];
|
|
BOOL needToReevaluate = [self.pendingWriteTree removeWriteId:writeId];
|
|
if (write.visible) {
|
|
if (persist) {
|
|
[self.persistenceManager removeUserWrite:writeId];
|
|
}
|
|
if (!revert) {
|
|
NSDictionary *serverValues = [FServerValues generateServerValues:clock];
|
|
if ([write isOverwrite]) {
|
|
id<FNode> resolvedNode = [FServerValues resolveDeferredValueSnapshot:write.overwrite withServerValues:serverValues];
|
|
[self.persistenceManager applyUserWrite:resolvedNode toServerCacheAtPath:write.path];
|
|
} else {
|
|
FCompoundWrite *resolvedMerge = [FServerValues resolveDeferredValueCompoundWrite:write.merge withServerValues:serverValues];
|
|
[self.persistenceManager applyUserMerge:resolvedMerge toServerCacheAtPath:write.path];
|
|
}
|
|
}
|
|
}
|
|
if (!needToReevaluate) {
|
|
return @[];
|
|
} else {
|
|
__block FImmutableTree *affectedTree = [FImmutableTree empty];
|
|
if (write.isOverwrite) {
|
|
affectedTree = [affectedTree setValue:@YES atPath:[FPath empty]];
|
|
} else {
|
|
[write.merge enumerateWrites:^(FPath *path, id <FNode> node, BOOL *stop) {
|
|
affectedTree = [affectedTree setValue:@YES atPath:path];
|
|
}];
|
|
}
|
|
FAckUserWrite *operation = [[FAckUserWrite alloc] initWithPath:write.path affectedTree:affectedTree revert:revert];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Apply new server data for the specified path
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyServerOverwriteAtPath:(FPath *)path newData:(id <FNode>)newData {
|
|
[self.persistenceManager updateServerCacheWithNode:newData forQuery:[FQuerySpec defaultQueryAtPath:path]];
|
|
FOverwrite *operation = [[FOverwrite alloc] initWithSource:[FOperationSource serverInstance] path:path snap:newData];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
|
|
/**
|
|
* Applied new server data to be merged in at the specified path
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyServerMergeAtPath:(FPath *)path changedChildren:(FCompoundWrite *)changedChildren {
|
|
[self.persistenceManager updateServerCacheWithMerge:changedChildren atPath:path];
|
|
FMerge *operation = [[FMerge alloc] initWithSource:[FOperationSource serverInstance] path:path children:changedChildren];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
|
|
- (NSArray *) applyServerRangeMergeAtPath:(FPath *)path updates:(NSArray *)ranges {
|
|
FSyncPoint *syncPoint = [self.syncPointTree valueAtPath:path];
|
|
if (syncPoint == nil) {
|
|
// Removed view, so it's safe to just ignore this update
|
|
return @[];
|
|
} else {
|
|
// This could be for any "complete" (unfiltered) view, and if there is more than one complete view, they should
|
|
// each have the same cache so it doesn't matter which one we use.
|
|
FView *view = [syncPoint completeView];
|
|
if (view != nil) {
|
|
id<FNode> serverNode = [view serverCache];
|
|
for (FRangeMerge *merge in ranges) {
|
|
serverNode = [merge applyToNode:serverNode];
|
|
}
|
|
return [self applyServerOverwriteAtPath:path newData:serverNode];
|
|
} else {
|
|
// There doesn't exist a view for this update, so it was removed and it's safe to just ignore this range
|
|
// merge
|
|
return @[];
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Apply a listen complete to a path
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyListenCompleteAtPath:(FPath *)path {
|
|
[self.persistenceManager setQueryComplete:[FQuerySpec defaultQueryAtPath:path]];
|
|
id<FOperation> operation = [[FListenComplete alloc] initWithSource:[FOperationSource serverInstance] path:path];
|
|
return [self applyOperationToSyncPoints:operation];
|
|
}
|
|
|
|
/**
|
|
* Apply a listen complete to a path
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyTaggedListenCompleteAtPath:(FPath *)path tagId:(NSNumber *)tagId {
|
|
FQuerySpec *query = [self queryForTag:tagId];
|
|
if (query != nil) {
|
|
[self.persistenceManager setQueryComplete:query];
|
|
FPath *relativePath = [FPath relativePathFrom:query.path to:path];
|
|
id<FOperation> op = [[FListenComplete alloc] initWithSource:[FOperationSource forServerTaggedQuery:query.params]
|
|
path:relativePath];
|
|
return [self applyTaggedOperation:op atPath:query.path];
|
|
} else {
|
|
// We've already removed the query. No big deal, ignore the update.
|
|
return @[];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Internal helper method to apply tagged operation
|
|
*/
|
|
- (NSArray *) applyTaggedOperation:(id<FOperation>)operation atPath:(FPath *)path {
|
|
FSyncPoint *syncPoint = [self.syncPointTree valueAtPath:path];
|
|
NSAssert(syncPoint != nil, @"Missing sync point for query tag that we're tracking.");
|
|
FWriteTreeRef *writesCache = [self.pendingWriteTree childWritesForPath:path];
|
|
return [syncPoint applyOperation:operation writesCache:writesCache serverCache:nil];
|
|
}
|
|
|
|
/**
|
|
* Apply new server data for the specified tagged query
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyTaggedQueryOverwriteAtPath:(FPath *)path newData:(id <FNode>)newData tagId:(NSNumber *)tagId {
|
|
FQuerySpec *query = [self queryForTag:tagId];
|
|
if (query != nil) {
|
|
FPath *relativePath = [FPath relativePathFrom:query.path to:path];
|
|
FQuerySpec *queryToOverwrite = relativePath.isEmpty ? query : [FQuerySpec defaultQueryAtPath:path];
|
|
[self.persistenceManager updateServerCacheWithNode:newData forQuery:queryToOverwrite];
|
|
FOverwrite *operation = [[FOverwrite alloc] initWithSource:[FOperationSource forServerTaggedQuery:query.params]
|
|
path:relativePath snap:newData];
|
|
return [self applyTaggedOperation:operation atPath:query.path];
|
|
} else {
|
|
// Query must have been removed already
|
|
return @[];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Apply server data to be merged in for the specified tagged query
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) applyTaggedQueryMergeAtPath:(FPath *)path changedChildren:(FCompoundWrite *)changedChildren tagId:(NSNumber *)tagId {
|
|
FQuerySpec *query = [self queryForTag:tagId];
|
|
if (query != nil) {
|
|
FPath *relativePath = [FPath relativePathFrom:query.path to:path];
|
|
[self.persistenceManager updateServerCacheWithMerge:changedChildren atPath:path];
|
|
FMerge *operation = [[FMerge alloc] initWithSource:[FOperationSource forServerTaggedQuery:query.params]
|
|
path:relativePath
|
|
children:changedChildren];
|
|
return [self applyTaggedOperation:operation atPath:query.path];
|
|
} else {
|
|
// We've already removed the query. No big deal, ignore the update.
|
|
return @[];
|
|
}
|
|
}
|
|
|
|
- (NSArray *) applyTaggedServerRangeMergeAtPath:(FPath *)path updates:(NSArray *)ranges tagId:(NSNumber *)tagId {
|
|
FQuerySpec *query = [self queryForTag:tagId];
|
|
if (query != nil) {
|
|
NSAssert([path isEqual:query.path], @"Tagged update path and query path must match");
|
|
FSyncPoint *syncPoint = [self.syncPointTree valueAtPath:path];
|
|
NSAssert(syncPoint != nil, @"Missing sync point for query tag that we're tracking.");
|
|
FView *view = [syncPoint viewForQuery:query];
|
|
NSAssert(view != nil, @"Missing view for query tag that we're tracking");
|
|
id<FNode> serverNode = [view serverCache];
|
|
for (FRangeMerge *merge in ranges) {
|
|
serverNode = [merge applyToNode:serverNode];
|
|
}
|
|
return [self applyTaggedQueryOverwriteAtPath:path newData:serverNode tagId:tagId];
|
|
} else {
|
|
// We've already removed the query. No big deal, ignore the update.
|
|
return @[];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Add an event callback for the specified query
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) addEventRegistration:(id<FEventRegistration>)eventRegistration forQuery:(FQuerySpec *)query {
|
|
FPath *path = query.path;
|
|
|
|
__block BOOL foundAncestorDefaultView = NO;
|
|
[self.syncPointTree forEachOnPath:query.path whileBlock:^BOOL(FPath *pathToSyncPoint, FSyncPoint *syncPoint) {
|
|
foundAncestorDefaultView = foundAncestorDefaultView || [syncPoint hasCompleteView];
|
|
return !foundAncestorDefaultView;
|
|
}];
|
|
|
|
[self.persistenceManager setQueryActive:query];
|
|
|
|
FSyncPoint *syncPoint = [self.syncPointTree valueAtPath:path];
|
|
if (syncPoint == nil) {
|
|
syncPoint = [[FSyncPoint alloc] initWithPersistenceManager:self.persistenceManager];
|
|
self.syncPointTree = [self.syncPointTree setValue:syncPoint atPath:path];
|
|
}
|
|
|
|
BOOL viewAlreadyExists = [syncPoint viewExistsForQuery:query];
|
|
NSArray *events;
|
|
if (viewAlreadyExists) {
|
|
events = [syncPoint addEventRegistration:eventRegistration forExistingViewForQuery:query];
|
|
} else {
|
|
if (![query loadsAllData]) {
|
|
// We need to track a tag for this query
|
|
NSAssert(self.queryToTagMap[query] == nil, @"View does not exist, but we have a tag");
|
|
NSNumber *tagId = [self.queryTagCounter getAndIncrement];
|
|
self.queryToTagMap[query] = tagId;
|
|
self.tagToQueryMap[tagId] = query;
|
|
}
|
|
|
|
FWriteTreeRef *writesCache = [self.pendingWriteTree childWritesForPath:path];
|
|
FCacheNode *serverCache = [self serverCacheForQuery:query];
|
|
events = [syncPoint addEventRegistration:eventRegistration
|
|
forNonExistingViewForQuery:query
|
|
writesCache:writesCache
|
|
serverCache:serverCache];
|
|
|
|
// There was no view and no default listen
|
|
if (!foundAncestorDefaultView) {
|
|
FView *view = [syncPoint viewForQuery:query];
|
|
NSMutableArray *mutableEvents = [events mutableCopy];
|
|
[mutableEvents addObjectsFromArray:[self setupListenerOnQuery:query view:view]];
|
|
events = mutableEvents;
|
|
}
|
|
}
|
|
|
|
return events;
|
|
}
|
|
|
|
- (FCacheNode *)serverCacheForQuery:(FQuerySpec *)query {
|
|
__block id<FNode> serverCacheNode = nil;
|
|
|
|
[self.syncPointTree forEachOnPath:query.path whileBlock:^BOOL(FPath *pathToSyncPoint, FSyncPoint *syncPoint) {
|
|
FPath *relativePath = [FPath relativePathFrom:pathToSyncPoint to:query.path];
|
|
serverCacheNode = [syncPoint completeServerCacheAtPath:relativePath];
|
|
return serverCacheNode == nil;
|
|
}];
|
|
|
|
FCacheNode *serverCache;
|
|
if (serverCacheNode != nil) {
|
|
FIndexedNode *indexed = [FIndexedNode indexedNodeWithNode:serverCacheNode index:query.index];
|
|
serverCache = [[FCacheNode alloc] initWithIndexedNode:indexed isFullyInitialized:YES isFiltered:NO];
|
|
} else {
|
|
FCacheNode *persistenceServerCache = [self.persistenceManager serverCacheForQuery:query];
|
|
if (persistenceServerCache.isFullyInitialized) {
|
|
serverCache = persistenceServerCache;
|
|
} else {
|
|
serverCacheNode = [FEmptyNode emptyNode];
|
|
|
|
FImmutableTree *subtree = [self.syncPointTree subtreeAtPath:query.path];
|
|
[subtree forEachChild:^(NSString *childKey, FSyncPoint *childSyncPoint) {
|
|
id<FNode> completeCache = [childSyncPoint completeServerCacheAtPath:[FPath empty]];
|
|
if (completeCache) {
|
|
serverCacheNode = [serverCacheNode updateImmediateChild:childKey withNewChild:completeCache];
|
|
}
|
|
}];
|
|
// Fill the node with any available children we have
|
|
[persistenceServerCache.node enumerateChildrenUsingBlock:^(NSString *key, id<FNode> node, BOOL *stop) {
|
|
if (![serverCacheNode hasChild:key]) {
|
|
serverCacheNode = [serverCacheNode updateImmediateChild:key withNewChild:node];
|
|
}
|
|
}];
|
|
FIndexedNode *indexed = [FIndexedNode indexedNodeWithNode:serverCacheNode index:query.index];
|
|
serverCache = [[FCacheNode alloc] initWithIndexedNode:indexed isFullyInitialized:NO isFiltered:NO];
|
|
}
|
|
}
|
|
|
|
return serverCache;
|
|
}
|
|
|
|
/**
|
|
* Remove event callback(s).
|
|
*
|
|
* If query is the default query, we'll check all queries for the specified eventRegistration.
|
|
* If eventRegistration is null, we'll remove all callbacks for the specified query/queries.
|
|
*
|
|
* @param eventRegistration if nil, all callbacks are removed
|
|
* @param cancelError If provided, appropriate cancel events will be returned
|
|
* @return NSArray of FEvent to raise.
|
|
*/
|
|
- (NSArray *) removeEventRegistration:(id <FEventRegistration>)eventRegistration
|
|
forQuery:(FQuerySpec *)query
|
|
cancelError:(NSError *)cancelError {
|
|
// Find the syncPoint first. Then deal with whether or not it has matching listeners
|
|
FPath *path = query.path;
|
|
FSyncPoint *maybeSyncPoint = [self.syncPointTree valueAtPath:path];
|
|
NSArray *cancelEvents = @[];
|
|
|
|
// A removal on a default query affects all queries at that location. A removal on an indexed query, even one without
|
|
// other query constraints, does *not* affect all queries at that location. So this check must be for 'default', and
|
|
// not loadsAllData:
|
|
if (maybeSyncPoint && ([query isDefault] || [maybeSyncPoint viewExistsForQuery:query])) {
|
|
FTupleRemovedQueriesEvents *removedAndEvents = [maybeSyncPoint removeEventRegistration:eventRegistration forQuery:query cancelError:cancelError];
|
|
if ([maybeSyncPoint isEmpty]) {
|
|
self.syncPointTree = [self.syncPointTree removeValueAtPath:path];
|
|
}
|
|
NSArray *removed = removedAndEvents.removedQueries;
|
|
cancelEvents = removedAndEvents.cancelEvents;
|
|
|
|
// We may have just removed one of many listeners and can short-circuit this whole process
|
|
// We may also not have removed a default listener, in which case all of the descendant listeners should already
|
|
// be properly set up.
|
|
//
|
|
// Since indexed queries can shadow if they don't have other query constraints, check for loadsAllData: instead
|
|
// of isDefault:
|
|
NSUInteger defaultQueryIndex = [removed indexOfObjectPassingTest:^BOOL(FQuerySpec *q, NSUInteger idx, BOOL *stop) {
|
|
return [q loadsAllData];
|
|
}];
|
|
BOOL removingDefault = defaultQueryIndex != NSNotFound;
|
|
[removed enumerateObjectsUsingBlock:^(FQuerySpec *query, NSUInteger idx, BOOL *stop) {
|
|
[self.persistenceManager setQueryInactive:query];
|
|
}];
|
|
NSNumber *covered = [self.syncPointTree findOnPath:path andApplyBlock:^id(FPath *relativePath, FSyncPoint *parentSyncPoint) {
|
|
return [NSNumber numberWithBool:[parentSyncPoint hasCompleteView]];
|
|
}];
|
|
|
|
if (removingDefault && ![covered boolValue]) {
|
|
FImmutableTree *subtree = [self.syncPointTree subtreeAtPath:path];
|
|
// There are potentially child listeners. Determine what if any listens we need to send before executing
|
|
// the removal
|
|
if (![subtree isEmpty]) {
|
|
// We need to fold over our subtree and collect the listeners to send
|
|
NSArray *newViews = [self collectDistinctViewsForSubTree:subtree];
|
|
|
|
// Ok, we've collected all the listens we need. Set them up.
|
|
[newViews enumerateObjectsUsingBlock:^(FView *view, NSUInteger idx, BOOL *stop) {
|
|
FQuerySpec *newQuery = view.query;
|
|
FListenContainer *listenContainer = [self createListenerForView:view];
|
|
self.listenProvider.startListening([self queryForListening:newQuery], [self tagForQuery:newQuery],
|
|
listenContainer, listenContainer.onComplete);
|
|
}];
|
|
} else {
|
|
// There's nothing below us, so nothing we need to start listening on
|
|
}
|
|
}
|
|
|
|
// If we removed anything and we're not covered by a higher up listen, we need to stop listening on this query.
|
|
// The above block has us covered in terms of making sure we're set up on listens lower in the tree.
|
|
// Also, note that if we have a cancelError, it's already been removed at the provider level.
|
|
if (![covered boolValue] && [removed count] > 0 && cancelError == nil) {
|
|
// If we removed a default, then we weren't listening on any of the other queries here. Just cancel the one
|
|
// default. Otherwise, we need to iterate through and cancel each individual query
|
|
if (removingDefault) {
|
|
// We don't tag default listeners
|
|
self.listenProvider.stopListening([self queryForListening:query], nil);
|
|
} else {
|
|
[removed enumerateObjectsUsingBlock:^(FQuerySpec *queryToRemove, NSUInteger idx, BOOL *stop) {
|
|
NSNumber *tagToRemove = [self.queryToTagMap objectForKey:queryToRemove];
|
|
self.listenProvider.stopListening([self queryForListening:queryToRemove], tagToRemove);
|
|
}];
|
|
}
|
|
}
|
|
// Now, clear all the tags we're tracking for the removed listens.
|
|
[self removeTags:removed];
|
|
} else {
|
|
// No-op, this listener must've been already removed
|
|
}
|
|
return cancelEvents;
|
|
}
|
|
|
|
- (void)keepQuery:(FQuerySpec *)query synced:(BOOL)keepSynced {
|
|
// Only do something if we actually need to add/remove an event registration
|
|
if (keepSynced && ![self.keepSyncedQueries containsObject:query]) {
|
|
[self addEventRegistration:[FKeepSyncedEventRegistration instance] forQuery:query];
|
|
[self.keepSyncedQueries addObject:query];
|
|
} else if (!keepSynced && [self.keepSyncedQueries containsObject:query]) {
|
|
[self removeEventRegistration:[FKeepSyncedEventRegistration instance] forQuery:query cancelError:nil];
|
|
[self.keepSyncedQueries removeObject:query];
|
|
}
|
|
}
|
|
|
|
- (NSArray *) removeAllWrites {
|
|
[self.persistenceManager removeAllUserWrites];
|
|
NSArray *removedWrites = [self.pendingWriteTree removeAllWrites];
|
|
if (removedWrites.count > 0) {
|
|
FImmutableTree *affectedTree = [[FImmutableTree empty] setValue:@YES atPath:[FPath empty]];
|
|
return [self applyOperationToSyncPoints:[[FAckUserWrite alloc] initWithPath:[FPath empty]
|
|
affectedTree:affectedTree revert:YES]];
|
|
} else {
|
|
return @[];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Returns a complete cache, if we have one, of the data at a particular path. The location must have a listener above
|
|
* it, but as this is only used by transaction code, that should always be the case anyways.
|
|
*
|
|
* Note: this method will *include* hidden writes from transaction with applyLocally set to false.
|
|
* @param path The path to the data we want
|
|
* @param writeIdsToExclude A specific set to be excluded
|
|
*/
|
|
- (id <FNode>) calcCompleteEventCacheAtPath:(FPath *)path excludeWriteIds:(NSArray *)writeIdsToExclude {
|
|
BOOL includeHiddenSets = YES;
|
|
FWriteTree *writeTree = self.pendingWriteTree;
|
|
id<FNode> serverCache = [self.syncPointTree findOnPath:path andApplyBlock:^id<FNode>(FPath *pathSoFar, FSyncPoint *syncPoint) {
|
|
FPath *relativePath = [FPath relativePathFrom:pathSoFar to:path];
|
|
id<FNode> serverCache = [syncPoint completeServerCacheAtPath:relativePath];
|
|
if (serverCache) {
|
|
return serverCache;
|
|
} else {
|
|
return nil;
|
|
}
|
|
}];
|
|
return [writeTree calculateCompleteEventCacheAtPath:path completeServerCache:serverCache excludeWriteIds:writeIdsToExclude includeHiddenWrites:includeHiddenSets];
|
|
}
|
|
|
|
#pragma mark -
|
|
#pragma mark Private Methods
|
|
/**
|
|
* This collapses multiple unfiltered views into a single view, since we only need a single
|
|
* listener for them.
|
|
* @return NSArray of FView
|
|
*/
|
|
- (NSArray *) collectDistinctViewsForSubTree:(FImmutableTree *)subtree {
|
|
return [subtree foldWithBlock:^NSArray *(FPath *relativePath, FSyncPoint *maybeChildSyncPoint, NSDictionary *childMap) {
|
|
if (maybeChildSyncPoint && [maybeChildSyncPoint hasCompleteView]) {
|
|
FView *completeView = [maybeChildSyncPoint completeView];
|
|
return @[completeView];
|
|
} else {
|
|
// No complete view here, flatten any deeper listens into an array
|
|
NSMutableArray *views = [[NSMutableArray alloc] init];
|
|
if (maybeChildSyncPoint) {
|
|
views = [[maybeChildSyncPoint queryViews] mutableCopy];
|
|
}
|
|
[childMap enumerateKeysAndObjectsUsingBlock:^(NSString *childKey, NSArray *childViews, BOOL *stop) {
|
|
[views addObjectsFromArray:childViews];
|
|
}];
|
|
return views;
|
|
}
|
|
}];
|
|
}
|
|
|
|
/**
|
|
* @param queries NSArray of FQuerySpec
|
|
*/
|
|
- (void) removeTags:(NSArray *)queries {
|
|
[queries enumerateObjectsUsingBlock:^(FQuerySpec *removedQuery, NSUInteger idx, BOOL *stop) {
|
|
if (![removedQuery loadsAllData]) {
|
|
// We should have a tag for this
|
|
NSNumber *removedQueryTag = self.queryToTagMap[removedQuery];
|
|
[self.queryToTagMap removeObjectForKey:removedQuery];
|
|
[self.tagToQueryMap removeObjectForKey:removedQueryTag];
|
|
}
|
|
}];
|
|
}
|
|
|
|
- (FQuerySpec *) queryForListening:(FQuerySpec *)query {
|
|
if (query.loadsAllData && !query.isDefault) {
|
|
// We treat queries that load all data as default queries
|
|
return [FQuerySpec defaultQueryAtPath:query.path];
|
|
} else {
|
|
return query;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* For a given new listen, manage the de-duplication of outstanding subscriptions.
|
|
* @return NSArray of FEvent events to support synchronous data sources
|
|
*/
|
|
- (NSArray *) setupListenerOnQuery:(FQuerySpec *)query view:(FView *)view {
|
|
FPath *path = query.path;
|
|
NSNumber *tagId = [self tagForQuery:query];
|
|
FListenContainer *listenContainer = [self createListenerForView:view];
|
|
|
|
NSArray *events = self.listenProvider.startListening([self queryForListening:query], tagId, listenContainer,
|
|
listenContainer.onComplete);
|
|
|
|
FImmutableTree *subtree = [self.syncPointTree subtreeAtPath:path];
|
|
// The root of this subtree has our query. We're here because we definitely need to send a listen for that, but we
|
|
// may need to shadow other listens as well.
|
|
if (tagId != nil) {
|
|
NSAssert(![subtree.value hasCompleteView], @"If we're adding a query, it shouldn't be shadowed");
|
|
} else {
|
|
// Shadow everything at or below this location, this is a default listener.
|
|
NSArray *queriesToStop = [subtree foldWithBlock:^id(FPath *relativePath, FSyncPoint *maybeChildSyncPoint, NSDictionary *childMap) {
|
|
if (![relativePath isEmpty] && maybeChildSyncPoint != nil && [maybeChildSyncPoint hasCompleteView]) {
|
|
return @[[maybeChildSyncPoint completeView].query];
|
|
} else {
|
|
// No default listener here, flatten any deeper queries into an array
|
|
NSMutableArray *queries = [[NSMutableArray alloc] init];
|
|
if (maybeChildSyncPoint != nil) {
|
|
for (FView *view in [maybeChildSyncPoint queryViews]) {
|
|
[queries addObject:view.query];
|
|
}
|
|
}
|
|
[childMap enumerateKeysAndObjectsUsingBlock:^(NSString *key, NSArray *childQueries, BOOL *stop) {
|
|
[queries addObjectsFromArray:childQueries];
|
|
}];
|
|
return queries;
|
|
}
|
|
}];
|
|
for (FQuerySpec *queryToStop in queriesToStop) {
|
|
self.listenProvider.stopListening([self queryForListening:queryToStop], [self tagForQuery:queryToStop]);
|
|
}
|
|
}
|
|
return events;
|
|
}
|
|
|
|
- (FListenContainer *) createListenerForView:(FView *)view {
|
|
FQuerySpec *query = view.query;
|
|
NSNumber *tagId = [self tagForQuery:query];
|
|
|
|
FListenContainer *listenContainer = [[FListenContainer alloc] initWithView:view
|
|
onComplete:^(NSString *status) {
|
|
if ([status isEqualToString:@"ok"]) {
|
|
if (tagId != nil) {
|
|
return [self applyTaggedListenCompleteAtPath:query.path tagId:tagId];
|
|
} else {
|
|
return [self applyListenCompleteAtPath:query.path];
|
|
}
|
|
} else {
|
|
// If a listen failed, kill all of the listeners here, not just the one that triggered the error.
|
|
// Note that this may need to be scoped to just this listener if we change permissions on filtered children
|
|
NSError *error = [FUtilities errorForStatus:status andReason:nil];
|
|
FFWarn(@"I-RDB038012", @"Listener at %@ failed: %@", query.path, status);
|
|
return [self removeEventRegistration:nil forQuery:query cancelError:error];
|
|
}
|
|
}];
|
|
|
|
return listenContainer;
|
|
}
|
|
|
|
/**
|
|
* @return The query associated with the given tag, if we have one
|
|
*/
|
|
- (FQuerySpec *) queryForTag:(NSNumber *)tagId {
|
|
return self.tagToQueryMap[tagId];
|
|
}
|
|
|
|
/**
|
|
* @return The tag associated with the given query
|
|
*/
|
|
- (NSNumber *) tagForQuery:(FQuerySpec *)query {
|
|
return self.queryToTagMap[query];
|
|
}
|
|
|
|
#pragma mark -
|
|
#pragma mark applyOperation Helpers
|
|
|
|
/**
|
|
* A helper method that visits all descendant and ancestor SyncPoints, applying the operation.
|
|
*
|
|
* NOTES:
|
|
* - Descendant SyncPoints will be visited first (since we raise events depth-first).
|
|
|
|
* - We call applyOperation: on each SyncPoint passing three things:
|
|
* 1. A version of the Operation that has been made relative to the SyncPoint location.
|
|
* 2. A WriteTreeRef of any writes we have cached at the SyncPoint location.
|
|
* 3. A snapshot Node with cached server data, if we have it.
|
|
|
|
* - We concatenate all of the events returned by each SyncPoint and return the result.
|
|
*
|
|
* @return Array of FEvent
|
|
*/
|
|
- (NSArray *) applyOperationToSyncPoints:(id<FOperation>)operation {
|
|
return [self applyOperationHelper:operation syncPointTree:self.syncPointTree serverCache:nil
|
|
writesCache:[self.pendingWriteTree childWritesForPath:[FPath empty]]];
|
|
}
|
|
|
|
/**
|
|
* Recursive helper for applyOperationToSyncPoints_
|
|
*/
|
|
- (NSArray *) applyOperationHelper:(id<FOperation>)operation syncPointTree:(FImmutableTree *)syncPointTree
|
|
serverCache:(id<FNode>)serverCache writesCache:(FWriteTreeRef *)writesCache {
|
|
if ([operation.path isEmpty]) {
|
|
return [self applyOperationDescendantsHelper:operation syncPointTree:syncPointTree serverCache:serverCache writesCache:writesCache];
|
|
} else {
|
|
FSyncPoint *syncPoint = syncPointTree.value;
|
|
|
|
// If we don't have cached server data, see if we can get it from this SyncPoint
|
|
if (serverCache == nil && syncPoint != nil) {
|
|
serverCache = [syncPoint completeServerCacheAtPath:[FPath empty]];
|
|
}
|
|
|
|
NSMutableArray *events = [[NSMutableArray alloc] init];
|
|
NSString *childKey = [operation.path getFront];
|
|
id<FOperation> childOperation = [operation operationForChild:childKey];
|
|
FImmutableTree *childTree = [syncPointTree.children get:childKey];
|
|
if (childTree != nil && childOperation != nil) {
|
|
id<FNode> childServerCache = serverCache ? [serverCache getImmediateChild:childKey] : nil;
|
|
FWriteTreeRef *childWritesCache = [writesCache childWriteTreeRef:childKey];
|
|
[events addObjectsFromArray:[self applyOperationHelper:childOperation syncPointTree:childTree serverCache:childServerCache writesCache:childWritesCache]];
|
|
}
|
|
|
|
if (syncPoint) {
|
|
[events addObjectsFromArray:[syncPoint applyOperation:operation writesCache:writesCache serverCache:serverCache]];
|
|
}
|
|
|
|
return events;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Recursive helper for applyOperationToSyncPoints:
|
|
*/
|
|
- (NSArray *) applyOperationDescendantsHelper:(id<FOperation>)operation syncPointTree:(FImmutableTree *)syncPointTree
|
|
serverCache:(id<FNode>)serverCache writesCache:(FWriteTreeRef *)writesCache {
|
|
FSyncPoint *syncPoint = syncPointTree.value;
|
|
|
|
// If we don't have cached server data, see if we can get it from this SyncPoint
|
|
id<FNode> resolvedServerCache;
|
|
if (serverCache == nil & syncPoint != nil) {
|
|
resolvedServerCache = [syncPoint completeServerCacheAtPath:[FPath empty]];
|
|
} else {
|
|
resolvedServerCache = serverCache;
|
|
}
|
|
|
|
NSMutableArray *events = [[NSMutableArray alloc] init];
|
|
[syncPointTree.children enumerateKeysAndObjectsUsingBlock:^(NSString *childKey, FImmutableTree *childTree, BOOL *stop) {
|
|
id<FNode> childServerCache = nil;
|
|
if (resolvedServerCache != nil) {
|
|
childServerCache = [resolvedServerCache getImmediateChild:childKey];
|
|
}
|
|
FWriteTreeRef *childWritesCache = [writesCache childWriteTreeRef:childKey];
|
|
id<FOperation> childOperation = [operation operationForChild:childKey];
|
|
if (childOperation != nil) {
|
|
[events addObjectsFromArray:[self applyOperationDescendantsHelper:childOperation
|
|
syncPointTree:childTree
|
|
serverCache:childServerCache
|
|
writesCache:childWritesCache]];
|
|
}
|
|
}];
|
|
|
|
if (syncPoint) {
|
|
[events addObjectsFromArray:[syncPoint applyOperation:operation writesCache:writesCache serverCache:resolvedServerCache]];
|
|
}
|
|
|
|
return events;
|
|
}
|
|
|
|
@end
|