| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936 |
12x
12x
12x
12x
12x
12x
12x
12x
12x
12x
12x
12x
12x
12x
86x
86x
86x
86x
86x
12x
524x
524x
23x
501x
12x
11x
11x
11x
595x
533x
533x
533x
108x
425x
425x
414x
11x
27x
425x
12x
278x
12x
2x
2x
12x
137x
12x
199x
199x
199x
199x
199x
199x
199x
199x
12x
9x
9x
9x
9x
9x
9x
9x
9x
9x
12x
217x
217x
217x
217x
217x
217x
217x
217x
12x
410x
410x
410x
410x
18x
18x
18x
410x
410x
256x
256x
154x
154x
410x
30x
380x
380x
380x
380x
9x
9x
2x
410x
410x
144x
144x
144x
144x
144x
410x
410x
410x
290x
290x
410x
160x
160x
160x
160x
160x
159x
159x
122x
159x
159x
159x
129x
159x
48x
159x
103x
103x
7x
7x
9x
9x
9x
9x
159x
119x
103x
103x
16x
16x
16x
159x
160x
12x
311x
311x
311x
326x
326x
326x
117x
311x
12x
7x
18x
8x
8x
10x
10x
1x
10x
11x
10x
12x
159x
129x
129x
21x
21x
21x
21x
12x
428x
11x
417x
12x
290x
290x
290x
290x
290x
290x
140x
150x
160x
7x
153x
153x
152x
3x
153x
10x
153x
150x
10x
10x
290x
299x
299x
299x
299x
396x
396x
354x
354x
217x
137x
12x
789x
12x
425x
425x
425x
12x
425x
12x
608x
608x
12x
12x
144x
12x
425x
425x
425x
425x
12x
1354x
12x
2381x
540x
1841x
1841x
301x
1841x
1841x
1841x
1841x
1841x
1027x
1027x
1027x
1841x
301x
1841x
581x
581x
581x
552x
581x
581x
43x
43x
43x
43x
41x
581x
577x
581x
12x
| /**
* Copyright 2017 Google Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { assert } from '@firebase/util';
import { errorForServerCode } from './util/util';
import { AckUserWrite } from './operation/AckUserWrite';
import { ChildrenNode } from './snap/ChildrenNode';
import { forEach, safeGet } from '@firebase/util';
import { ImmutableTree } from './util/ImmutableTree';
import { ListenComplete } from './operation/ListenComplete';
import { Merge } from './operation/Merge';
import { Operation, OperationSource } from './operation/Operation';
import { Overwrite } from './operation/Overwrite';
import { Path } from './util/Path';
import { SyncPoint } from './SyncPoint';
import { WriteTree, WriteTreeRef } from './WriteTree';
import { Query } from '../api/Query';
import { Node } from './snap/Node';
import { Event } from './view/Event';
import { EventRegistration } from './view/EventRegistration';
import { View } from './view/View';
/**
* @typedef {{
* startListening: function(
* !Query,
* ?number,
* function():string,
* function(!string, *):!Array.<!Event>
* ):!Array.<!Event>,
*
* stopListening: function(!Query, ?number)
* }}
*/
export interface ListenProvider {
startListening(
query: Query,
tag: number | null,
hashFn: () => string,
onComplete: (a: string, b?: any) => Event[]
): Event[];
stopListening(a: Query, b: number | null): void;
}
/**
* 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 set(), transaction(), and update() calls
* (applyUserOverwrite(), applyUserMerge()).
* - Applying and caching data changes for server data changes (applyServerOverwrite(),
* applyServerMerge()).
* - 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.
*
* @constructor
*/
export class SyncTree {
/**
* Tree of SyncPoints. There's a SyncPoint at any location that has 1 or more views.
* @type {!ImmutableTree.<!SyncPoint>}
* @private
*/
private syncPointTree_: ImmutableTree<SyncPoint> = ImmutableTree.Empty;
/**
* A tree of all pending user writes (user-initiated set()'s, transaction()'s, update()'s, etc.).
* @type {!WriteTree}
* @private
*/
private pendingWriteTree_ = new WriteTree();
private tagToQueryMap_: { [k: string]: string } = {};
private queryToTagMap_: { [k: string]: number } = {};
/**
* @param {!ListenProvider} listenProvider_ Used by SyncTree to start / stop listening
* to server data.
*/
constructor(private listenProvider_: ListenProvider) {}
/**
* Apply the data changes for a user-generated set() or transaction() call.
*
* @param {!Path} path
* @param {!Node} newData
* @param {number} writeId
* @param {boolean=} visible
* @return {!Array.<!Event>} Events to raise.
*/
applyUserOverwrite(
path: Path,
newData: Node,
writeId: number,
visible?: boolean
): Event[] {
// Record pending write.
this.pendingWriteTree_.addOverwrite(path, newData, writeId, visible);
if (!visible) {
return [];
} else {
return this.applyOperationToSyncPoints_(
new Overwrite(OperationSource.User, path, newData)
);
}
}
/**
* Apply the data from a user-generated update() call
*
* @param {!Path} path
* @param {!Object.<string, !Node>} changedChildren
* @param {!number} writeId
* @return {!Array.<!Event>} Events to raise.
*/
applyUserMerge(
path: Path,
changedChildren: { [k: string]: Node },
writeId: number
): Event[] {
// Record pending merge.
this.pendingWriteTree_.addMerge(path, changedChildren, writeId);
const changeTree = ImmutableTree.fromObject(changedChildren);
return this.applyOperationToSyncPoints_(
new Merge(OperationSource.User, path, changeTree)
);
}
/**
* Acknowledge a pending user write that was previously registered with applyUserOverwrite() or applyUserMerge().
*
* @param {!number} writeId
* @param {boolean=} revert True if the given write failed and needs to be reverted
* @return {!Array.<!Event>} Events to raise.
*/
ackUserWrite(writeId: number, revert: boolean = false) {
const write = this.pendingWriteTree_.getWrite(writeId);
const needToReevaluate = this.pendingWriteTree_.removeWrite(writeId);
if (!needToReevaluate) {
return [];
} else {
let affectedTree = ImmutableTree.Empty;
if (write.snap != null) {
// overwrite
affectedTree = affectedTree.set(Path.Empty, true);
} else {
forEach(write.children, function(pathString: string, node: Node) {
affectedTree = affectedTree.set(new Path(pathString), node);
});
}
return this.applyOperationToSyncPoints_(
new AckUserWrite(write.path, affectedTree, revert)
);
}
}
/**
* Apply new server data for the specified path..
*
* @param {!Path} path
* @param {!Node} newData
* @return {!Array.<!Event>} Events to raise.
*/
applyServerOverwrite(path: Path, newData: Node): Event[] {
return this.applyOperationToSyncPoints_(
new Overwrite(OperationSource.Server, path, newData)
);
}
/**
* Apply new server data to be merged in at the specified path.
*
* @param {!Path} path
* @param {!Object.<string, !Node>} changedChildren
* @return {!Array.<!Event>} Events to raise.
*/
applyServerMerge(
path: Path,
changedChildren: { [k: string]: Node }
): Event[] {
const changeTree = ImmutableTree.fromObject(changedChildren);
return this.applyOperationToSyncPoints_(
new Merge(OperationSource.Server, path, changeTree)
);
}
/**
* Apply a listen complete for a query
*
* @param {!Path} path
* @return {!Array.<!Event>} Events to raise.
*/
applyListenComplete(path: Path): Event[] {
return this.applyOperationToSyncPoints_(
new ListenComplete(OperationSource.Server, path)
);
}
/**
* Apply new server data for the specified tagged query.
*
* @param {!Path} path
* @param {!Node} snap
* @param {!number} tag
* @return {!Array.<!Event>} Events to raise.
*/
applyTaggedQueryOverwrite(path: Path, snap: Node, tag: number): Event[] {
const queryKey = this.queryKeyForTag_(tag);
Eif (queryKey != null) {
const r = SyncTree.parseQueryKey_(queryKey);
const queryPath = r.path,
queryId = r.queryId;
const relativePath = Path.relativePath(queryPath, path);
const op = new Overwrite(
OperationSource.forServerTaggedQuery(queryId),
relativePath,
snap
);
return this.applyTaggedOperation_(queryPath, op);
} else {
// Query must have been removed already
return [];
}
}
/**
* Apply server data to be merged in for the specified tagged query.
*
* @param {!Path} path
* @param {!Object.<string, !Node>} changedChildren
* @param {!number} tag
* @return {!Array.<!Event>} Events to raise.
*/
applyTaggedQueryMerge(
path: Path,
changedChildren: { [k: string]: Node },
tag: number
): Event[] {
const queryKey = this.queryKeyForTag_(tag);
Eif (queryKey) {
const r = SyncTree.parseQueryKey_(queryKey);
const queryPath = r.path,
queryId = r.queryId;
const relativePath = Path.relativePath(queryPath, path);
const changeTree = ImmutableTree.fromObject(changedChildren);
const op = new Merge(
OperationSource.forServerTaggedQuery(queryId),
relativePath,
changeTree
);
return this.applyTaggedOperation_(queryPath, op);
} else {
// We've already removed the query. No big deal, ignore the update
return [];
}
}
/**
* Apply a listen complete for a tagged query
*
* @param {!Path} path
* @param {!number} tag
* @return {!Array.<!Event>} Events to raise.
*/
applyTaggedListenComplete(path: Path, tag: number): Event[] {
const queryKey = this.queryKeyForTag_(tag);
Eif (queryKey) {
const r = SyncTree.parseQueryKey_(queryKey);
const queryPath = r.path,
queryId = r.queryId;
const relativePath = Path.relativePath(queryPath, path);
const op = new ListenComplete(
OperationSource.forServerTaggedQuery(queryId),
relativePath
);
return this.applyTaggedOperation_(queryPath, op);
} else {
// We've already removed the query. No big deal, ignore the update
return [];
}
}
/**
* Add an event callback for the specified query.
*
* @param {!Query} query
* @param {!EventRegistration} eventRegistration
* @return {!Array.<!Event>} Events to raise.
*/
addEventRegistration(
query: Query,
eventRegistration: EventRegistration
): Event[] {
const path = query.path;
let serverCache: Node | null = null;
let foundAncestorDefaultView = false;
// Any covering writes will necessarily be at the root, so really all we need to find is the server cache.
// Consider optimizing this once there's a better understanding of what actual behavior will be.
this.syncPointTree_.foreachOnPath(path, function(pathToSyncPoint, sp) {
const relativePath = Path.relativePath(pathToSyncPoint, path);
serverCache = serverCache || sp.getCompleteServerCache(relativePath);
foundAncestorDefaultView =
foundAncestorDefaultView || sp.hasCompleteView();
});
let syncPoint = this.syncPointTree_.get(path);
if (!syncPoint) {
syncPoint = new SyncPoint();
this.syncPointTree_ = this.syncPointTree_.set(path, syncPoint);
} else {
foundAncestorDefaultView =
foundAncestorDefaultView || syncPoint.hasCompleteView();
serverCache = serverCache || syncPoint.getCompleteServerCache(Path.Empty);
}
let serverCacheComplete;
if (serverCache != null) {
serverCacheComplete = true;
} else {
serverCacheComplete = false;
serverCache = ChildrenNode.EMPTY_NODE;
const subtree = this.syncPointTree_.subtree(path);
subtree.foreachChild(function(childName, childSyncPoint) {
const completeCache = childSyncPoint.getCompleteServerCache(Path.Empty);
if (completeCache) {
serverCache = serverCache.updateImmediateChild(
childName,
completeCache
);
}
});
}
const viewAlreadyExists = syncPoint.viewExistsForQuery(query);
if (!viewAlreadyExists && !query.getQueryParams().loadsAllData()) {
// We need to track a tag for this query
const queryKey = SyncTree.makeQueryKey_(query);
assert(
!(queryKey in this.queryToTagMap_),
'View does not exist, but we have a tag'
);
const tag = SyncTree.getNextQueryTag_();
this.queryToTagMap_[queryKey] = tag;
// Coerce to string to avoid sparse arrays.
this.tagToQueryMap_['_' + tag] = queryKey;
}
const writesCache = this.pendingWriteTree_.childWrites(path);
let events = syncPoint.addEventRegistration(
query,
eventRegistration,
writesCache,
serverCache,
serverCacheComplete
);
if (!viewAlreadyExists && !foundAncestorDefaultView) {
const view /** @type !View */ = syncPoint.viewForQuery(query);
events = events.concat(this.setupListener_(query, view));
}
return events;
}
/**
* 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 {!Query} query
* @param {?EventRegistration} eventRegistration If null, all callbacks are removed.
* @param {Error=} cancelError If a cancelError is provided, appropriate cancel events will be returned.
* @return {!Array.<!Event>} Cancel events, if cancelError was provided.
*/
removeEventRegistration(
query: Query,
eventRegistration: EventRegistration | null,
cancelError?: Error
): Event[] {
// Find the syncPoint first. Then deal with whether or not it has matching listeners
const path = query.path;
const maybeSyncPoint = this.syncPointTree_.get(path);
let cancelEvents: Event[] = [];
// 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.queryIdentifier() === 'default' ||
maybeSyncPoint.viewExistsForQuery(query))
) {
/**
* @type {{removed: !Array.<!Query>, events: !Array.<!Event>}}
*/
const removedAndEvents = maybeSyncPoint.removeEventRegistration(
query,
eventRegistration,
cancelError
);
if (maybeSyncPoint.isEmpty()) {
this.syncPointTree_ = this.syncPointTree_.remove(path);
}
const removed = removedAndEvents.removed;
cancelEvents = removedAndEvents.events;
// 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
// queryId === 'default'
const removingDefault =
-1 !==
removed.findIndex(function(query) {
return query.getQueryParams().loadsAllData();
});
const covered = this.syncPointTree_.findOnPath(path, function(
relativePath,
parentSyncPoint
) {
return parentSyncPoint.hasCompleteView();
});
if (removingDefault && !covered) {
const subtree = this.syncPointTree_.subtree(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
const newViews = this.collectDistinctViewsForSubTree_(subtree);
// Ok, we've collected all the listens we need. Set them up.
for (let i = 0; i < newViews.length; ++i) {
const view = newViews[i],
newQuery = view.getQuery();
const listener = this.createListenerForView_(view);
this.listenProvider_.startListening(
SyncTree.queryForListening_(newQuery),
this.tagForQuery_(newQuery),
listener.hashFn,
listener.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 && removed.length > 0 && !cancelError) {
// 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
const defaultTag: number | null = null;
this.listenProvider_.stopListening(
SyncTree.queryForListening_(query),
defaultTag
);
} else {
removed.forEach((queryToRemove: Query) => {
const tagToRemove = this.queryToTagMap_[
SyncTree.makeQueryKey_(queryToRemove)
];
this.listenProvider_.stopListening(
SyncTree.queryForListening_(queryToRemove),
tagToRemove
);
});
}
}
// Now, clear all of the tags we're tracking for the removed listens
this.removeTags_(removed);
} else {
// No-op, this listener must've been already removed
}
return cancelEvents;
}
/**
* 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} path The path to the data we want
* @param {Array.<number>=} writeIdsToExclude A specific set to be excluded
* @return {?Node}
*/
calcCompleteEventCache(
path: Path,
writeIdsToExclude?: number[]
): Node | null {
const includeHiddenSets = true;
const writeTree = this.pendingWriteTree_;
const serverCache = this.syncPointTree_.findOnPath(path, function(
pathSoFar,
syncPoint
) {
const relativePath = Path.relativePath(pathSoFar, path);
const serverCache = syncPoint.getCompleteServerCache(relativePath);
if (serverCache) {
return serverCache;
}
});
return writeTree.calcCompleteEventCache(
path,
serverCache,
writeIdsToExclude,
includeHiddenSets
);
}
/**
* This collapses multiple unfiltered views into a single view, since we only need a single
* listener for them.
*
* @param {!ImmutableTree.<!SyncPoint>} subtree
* @return {!Array.<!View>}
* @private
*/
private collectDistinctViewsForSubTree_(
subtree: ImmutableTree<SyncPoint>
): View[] {
return subtree.fold<View[]>(
(relativePath, maybeChildSyncPoint, childMap) => {
if (maybeChildSyncPoint && maybeChildSyncPoint.hasCompleteView()) {
const completeView = maybeChildSyncPoint.getCompleteView();
return [completeView];
} else {
// No complete view here, flatten any deeper listens into an array
let views: View[] = [];
if (maybeChildSyncPoint) {
views = maybeChildSyncPoint.getQueryViews();
}
forEach(childMap, function(key: string, childViews: View[]) {
views = views.concat(childViews);
});
return views;
}
}
);
}
/**
* @param {!Array.<!Query>} queries
* @private
*/
private removeTags_(queries: Query[]) {
for (let j = 0; j < queries.length; ++j) {
const removedQuery = queries[j];
if (!removedQuery.getQueryParams().loadsAllData()) {
// We should have a tag for this
const removedQueryKey = SyncTree.makeQueryKey_(removedQuery);
const removedQueryTag = this.queryToTagMap_[removedQueryKey];
delete this.queryToTagMap_[removedQueryKey];
delete this.tagToQueryMap_['_' + removedQueryTag];
}
}
}
/**
* Normalizes a query to a query we send the server for listening
* @param {!Query} query
* @return {!Query} The normalized query
* @private
*/
private static queryForListening_(query: Query): Query {
if (
query.getQueryParams().loadsAllData() &&
!query.getQueryParams().isDefault()
) {
// We treat queries that load all data as default queries
// Cast is necessary because ref() technically returns Firebase which is actually fb.api.Firebase which inherits
// from Query
return /** @type {!Query} */ query.getRef();
} else {
return query;
}
}
/**
* For a given new listen, manage the de-duplication of outstanding subscriptions.
*
* @param {!Query} query
* @param {!View} view
* @return {!Array.<!Event>} This method can return events to support synchronous data sources
* @private
*/
private setupListener_(query: Query, view: View): Event[] {
const path = query.path;
const tag = this.tagForQuery_(query);
const listener = this.createListenerForView_(view);
const events = this.listenProvider_.startListening(
SyncTree.queryForListening_(query),
tag,
listener.hashFn,
listener.onComplete
);
const subtree = this.syncPointTree_.subtree(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 (tag) {
assert(
!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.
const queriesToStop = subtree.fold<Query[]>(function(
relativePath,
maybeChildSyncPoint,
childMap
) {
if (
!relativePath.isEmpty() &&
maybeChildSyncPoint &&
maybeChildSyncPoint.hasCompleteView()
) {
return [maybeChildSyncPoint.getCompleteView().getQuery()];
} else {
// No default listener here, flatten any deeper queries into an array
let queries: Query[] = [];
if (maybeChildSyncPoint) {
queries = queries.concat(
maybeChildSyncPoint.getQueryViews().map(view => view.getQuery())
);
}
forEach(childMap, function(key: string, childQueries: Query[]) {
queries = queries.concat(childQueries);
});
return queries;
}
});
for (let i = 0; i < queriesToStop.length; ++i) {
const queryToStop = queriesToStop[i];
this.listenProvider_.stopListening(
SyncTree.queryForListening_(queryToStop),
this.tagForQuery_(queryToStop)
);
}
}
return events;
}
/**
*
* @param {!View} view
* @return {{hashFn: function(), onComplete: function(!string, *)}}
* @private
*/
private createListenerForView_(
view: View
): { hashFn(): string; onComplete(a: string, b?: any): Event[] } {
const query = view.getQuery();
const tag = this.tagForQuery_(query);
return {
hashFn: () => {
const cache = view.getServerCache() || ChildrenNode.EMPTY_NODE;
return cache.hash();
},
onComplete: (status: string): Event[] => {
Eif (status === 'ok') {
if (tag) {
return this.applyTaggedListenComplete(query.path, tag);
} else {
return this.applyListenComplete(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
const error = errorForServerCode(status, query);
return this.removeEventRegistration(
query,
/*eventRegistration*/ null,
error
);
}
}
};
}
/**
* Given a query, computes a "queryKey" suitable for use in our queryToTagMap_.
* @private
* @param {!Query} query
* @return {string}
*/
private static makeQueryKey_(query: Query): string {
return query.path.toString() + '$' + query.queryIdentifier();
}
/**
* Given a queryKey (created by makeQueryKey), parse it back into a path and queryId.
* @private
* @param {!string} queryKey
* @return {{queryId: !string, path: !Path}}
*/
private static parseQueryKey_(
queryKey: string
): { queryId: string; path: Path } {
const splitIndex = queryKey.indexOf('$');
assert(
splitIndex !== -1 && splitIndex < queryKey.length - 1,
'Bad queryKey.'
);
return {
queryId: queryKey.substr(splitIndex + 1),
path: new Path(queryKey.substr(0, splitIndex))
};
}
/**
* Return the query associated with the given tag, if we have one
* @param {!number} tag
* @return {?string}
* @private
*/
private queryKeyForTag_(tag: number): string | null {
return this.tagToQueryMap_['_' + tag];
}
/**
* Return the tag associated with the given query.
* @param {!Query} query
* @return {?number}
* @private
*/
private tagForQuery_(query: Query): number | null {
const queryKey = SyncTree.makeQueryKey_(query);
return safeGet(this.queryToTagMap_, queryKey);
}
/**
* Static tracker for next query tag.
* @type {number}
* @private
*/
private static nextQueryTag_ = 1;
/**
* Static accessor for query tags.
* @return {number}
* @private
*/
private static getNextQueryTag_(): number {
return SyncTree.nextQueryTag_++;
}
/**
* A helper method to apply tagged operations
*
* @param {!Path} queryPath
* @param {!Operation} operation
* @return {!Array.<!Event>}
* @private
*/
private applyTaggedOperation_(
queryPath: Path,
operation: Operation
): Event[] {
const syncPoint = this.syncPointTree_.get(queryPath);
assert(syncPoint, "Missing sync point for query tag that we're tracking");
const writesCache = this.pendingWriteTree_.childWrites(queryPath);
return syncPoint.applyOperation(
operation,
writesCache,
/*serverCache=*/ null
);
}
/**
* 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.
*
* @param {!Operation} operation
* @return {!Array.<!Event>}
* @private
*/
private applyOperationToSyncPoints_(operation: Operation): Event[] {
return this.applyOperationHelper_(
operation,
this.syncPointTree_,
/*serverCache=*/ null,
this.pendingWriteTree_.childWrites(Path.Empty)
);
}
/**
* Recursive helper for applyOperationToSyncPoints_
*
* @private
* @param {!Operation} operation
* @param {ImmutableTree.<!SyncPoint>} syncPointTree
* @param {?Node} serverCache
* @param {!WriteTreeRef} writesCache
* @return {!Array.<!Event>}
*/
private applyOperationHelper_(
operation: Operation,
syncPointTree: ImmutableTree<SyncPoint>,
serverCache: Node | null,
writesCache: WriteTreeRef
): Event[] {
if (operation.path.isEmpty()) {
return this.applyOperationDescendantsHelper_(
operation,
syncPointTree,
serverCache,
writesCache
);
} else {
const syncPoint = syncPointTree.get(Path.Empty);
// If we don't have cached server data, see if we can get it from this SyncPoint.
if (serverCache == null && syncPoint != null) {
serverCache = syncPoint.getCompleteServerCache(Path.Empty);
}
let events: Event[] = [];
const childName = operation.path.getFront();
const childOperation = operation.operationForChild(childName);
const childTree = syncPointTree.children.get(childName);
if (childTree && childOperation) {
const childServerCache = serverCache
? serverCache.getImmediateChild(childName)
: null;
const childWritesCache = writesCache.child(childName);
events = events.concat(
this.applyOperationHelper_(
childOperation,
childTree,
childServerCache,
childWritesCache
)
);
}
if (syncPoint) {
events = events.concat(
syncPoint.applyOperation(operation, writesCache, serverCache)
);
}
return events;
}
}
/**
* Recursive helper for applyOperationToSyncPoints_
*
* @private
* @param {!Operation} operation
* @param {ImmutableTree.<!SyncPoint>} syncPointTree
* @param {?Node} serverCache
* @param {!WriteTreeRef} writesCache
* @return {!Array.<!Event>}
*/
private applyOperationDescendantsHelper_(
operation: Operation,
syncPointTree: ImmutableTree<SyncPoint>,
serverCache: Node | null,
writesCache: WriteTreeRef
): Event[] {
const syncPoint = syncPointTree.get(Path.Empty);
// If we don't have cached server data, see if we can get it from this SyncPoint.
if (serverCache == null && syncPoint != null) {
serverCache = syncPoint.getCompleteServerCache(Path.Empty);
}
let events: Event[] = [];
syncPointTree.children.inorderTraversal((childName, childTree) => {
const childServerCache = serverCache
? serverCache.getImmediateChild(childName)
: null;
const childWritesCache = writesCache.child(childName);
const childOperation = operation.operationForChild(childName);
if (childOperation) {
events = events.concat(
this.applyOperationDescendantsHelper_(
childOperation,
childTree,
childServerCache,
childWritesCache
)
);
}
});
if (syncPoint) {
events = events.concat(
syncPoint.applyOperation(operation, writesCache, serverCache)
);
}
return events;
}
}
|