FRepo.m 63 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520
  1. /*
  2. * Copyright 2017 Google
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #import <Foundation/Foundation.h>
  17. #import "FirebaseCore/Sources/Private/FirebaseCoreInternal.h"
  18. #import "FirebaseDatabase/Sources/Api/Private/FIRDataSnapshot_Private.h"
  19. #import "FirebaseDatabase/Sources/Api/Private/FIRDatabaseQuery_Private.h"
  20. #import "FirebaseDatabase/Sources/Api/Private/FIRDatabase_Private.h"
  21. #import "FirebaseDatabase/Sources/Api/Private/FIRMutableData_Private.h"
  22. #import "FirebaseDatabase/Sources/Api/Private/FIRTransactionResult_Private.h"
  23. #import "FirebaseDatabase/Sources/Constants/FConstants.h"
  24. #import "FirebaseDatabase/Sources/Core/FListenProvider.h"
  25. #import "FirebaseDatabase/Sources/Core/FQuerySpec.h"
  26. #import "FirebaseDatabase/Sources/Core/FRepo.h"
  27. #import "FirebaseDatabase/Sources/Core/FRepoManager.h"
  28. #import "FirebaseDatabase/Sources/Core/FRepo_Private.h"
  29. #import "FirebaseDatabase/Sources/Core/FServerValues.h"
  30. #import "FirebaseDatabase/Sources/Core/FSnapshotHolder.h"
  31. #import "FirebaseDatabase/Sources/Core/FSyncTree.h"
  32. #import "FirebaseDatabase/Sources/Core/FWriteRecord.h"
  33. #import "FirebaseDatabase/Sources/Core/Utilities/FTree.h"
  34. #import "FirebaseDatabase/Sources/Core/View/FEventRaiser.h"
  35. #import "FirebaseDatabase/Sources/Core/View/FEventRegistration.h"
  36. #import "FirebaseDatabase/Sources/Core/View/FValueEventRegistration.h"
  37. #import "FirebaseDatabase/Sources/FClock.h"
  38. #import "FirebaseDatabase/Sources/FIRDatabaseConfig_Private.h"
  39. #import "FirebaseDatabase/Sources/Persistence/FCachePolicy.h"
  40. #import "FirebaseDatabase/Sources/Persistence/FLevelDBStorageEngine.h"
  41. #import "FirebaseDatabase/Sources/Persistence/FPersistenceManager.h"
  42. #import "FirebaseDatabase/Sources/Public/FirebaseDatabase/FIRDataSnapshot.h"
  43. #import "FirebaseDatabase/Sources/Public/FirebaseDatabase/FIRMutableData.h"
  44. #import "FirebaseDatabase/Sources/Public/FirebaseDatabase/FIRTransactionResult.h"
  45. #import "FirebaseDatabase/Sources/Snapshot/FEmptyNode.h"
  46. #import "FirebaseDatabase/Sources/Snapshot/FSnapshotUtilities.h"
  47. #import "FirebaseDatabase/Sources/Utilities/FAtomicNumber.h"
  48. #import "FirebaseDatabase/Sources/Utilities/Tuples/FTupleNodePath.h"
  49. #import "FirebaseDatabase/Sources/Utilities/Tuples/FTupleSetIdPath.h"
  50. #import "FirebaseDatabase/Sources/Utilities/Tuples/FTupleTransaction.h"
  51. #import <dlfcn.h>
  52. #if TARGET_OS_IOS || TARGET_OS_TV
  53. #import <UIKit/UIKit.h>
  54. #endif
  55. @interface FRepo ()
  56. @property(nonatomic, strong) FOffsetClock *serverClock;
  57. @property(nonatomic, strong) FPersistenceManager *persistenceManager;
  58. @property(nonatomic, strong) FIRDatabase *database;
  59. @property(nonatomic, strong, readwrite) FAuthenticationManager *auth;
  60. @property(nonatomic, strong) FSyncTree *infoSyncTree;
  61. @property(nonatomic) NSInteger writeIdCounter;
  62. @property(nonatomic) BOOL hijackHash;
  63. @property(nonatomic, strong) FTree *transactionQueueTree;
  64. @property(nonatomic) BOOL loggedTransactionPersistenceWarning;
  65. /**
  66. * Test only. For load testing the server.
  67. */
  68. @property(nonatomic, strong) id (^interceptServerDataCallback)
  69. (NSString *pathString, id data);
  70. @end
  71. @implementation FRepo
  72. - (id)initWithRepoInfo:(FRepoInfo *)info
  73. config:(FIRDatabaseConfig *)config
  74. database:(FIRDatabase *)database {
  75. self = [super init];
  76. if (self) {
  77. self.repoInfo = info;
  78. self.config = config;
  79. self.database = database;
  80. // Access can occur outside of shared queue, so the clock needs to be
  81. // initialized here
  82. self.serverClock =
  83. [[FOffsetClock alloc] initWithClock:[FSystemClock clock] offset:0];
  84. self.connection = [[FPersistentConnection alloc]
  85. initWithRepoInfo:self.repoInfo
  86. dispatchQueue:[FIRDatabaseQuery sharedQueue]
  87. config:self.config];
  88. // Needs to be called before authentication manager is instantiated
  89. self.eventRaiser =
  90. [[FEventRaiser alloc] initWithQueue:self.config.callbackQueue];
  91. dispatch_async([FIRDatabaseQuery sharedQueue], ^{
  92. [self deferredInit];
  93. });
  94. }
  95. return self;
  96. }
  97. - (void)deferredInit {
  98. // TODO: cleanup on dealloc
  99. __weak FRepo *weakSelf = self;
  100. [self.config.authTokenProvider listenForTokenChanges:^(NSString *token) {
  101. [weakSelf.connection refreshAuthToken:token];
  102. }];
  103. // Open connection now so that by the time we are connected the deferred
  104. // init has run This relies on the fact that all callbacks run on repos
  105. // queue
  106. self.connection.delegate = self;
  107. [self.connection open];
  108. self.dataUpdateCount = 0;
  109. self.rangeMergeUpdateCount = 0;
  110. self.interceptServerDataCallback = nil;
  111. if (self.config.persistenceEnabled) {
  112. NSString *repoHashString =
  113. [NSString stringWithFormat:@"%@_%@", self.repoInfo.host,
  114. self.repoInfo.namespace];
  115. NSString *persistencePrefix =
  116. [NSString stringWithFormat:@"%@/%@", self.config.sessionIdentifier,
  117. repoHashString];
  118. id<FCachePolicy> cachePolicy = [[FLRUCachePolicy alloc]
  119. initWithMaxSize:self.config.persistenceCacheSizeBytes];
  120. id<FStorageEngine> engine;
  121. if (self.config.forceStorageEngine != nil) {
  122. engine = self.config.forceStorageEngine;
  123. } else {
  124. FLevelDBStorageEngine *levelDBEngine =
  125. [[FLevelDBStorageEngine alloc] initWithPath:persistencePrefix];
  126. // We need the repo info to run the legacy migration. Future
  127. // migrations will be managed by the database itself Remove this
  128. // once we are confident that no-one is using legacy migration
  129. // anymore...
  130. [levelDBEngine runLegacyMigration:self.repoInfo];
  131. engine = levelDBEngine;
  132. }
  133. self.persistenceManager =
  134. [[FPersistenceManager alloc] initWithStorageEngine:engine
  135. cachePolicy:cachePolicy];
  136. } else {
  137. self.persistenceManager = nil;
  138. }
  139. [self initTransactions];
  140. // A list of data pieces and paths to be set when this client disconnects
  141. self.onDisconnect = [[FSparseSnapshotTree alloc] init];
  142. self.infoData = [[FSnapshotHolder alloc] init];
  143. FListenProvider *infoListenProvider = [[FListenProvider alloc] init];
  144. infoListenProvider.startListening =
  145. ^(FQuerySpec *query, NSNumber *tagId, id<FSyncTreeHash> hash,
  146. fbt_nsarray_nsstring onComplete) {
  147. NSArray *infoEvents = @[];
  148. FRepo *strongSelf = weakSelf;
  149. id<FNode> node = [strongSelf.infoData getNode:query.path];
  150. // This is possibly a hack, but we have different semantics for .info
  151. // endpoints. We don't raise null events on initial data...
  152. if (![node isEmpty]) {
  153. infoEvents =
  154. [strongSelf.infoSyncTree applyServerOverwriteAtPath:query.path
  155. newData:node];
  156. [strongSelf.eventRaiser raiseCallback:^{
  157. onComplete(kFWPResponseForActionStatusOk);
  158. }];
  159. }
  160. return infoEvents;
  161. };
  162. infoListenProvider.stopListening = ^(FQuerySpec *query, NSNumber *tagId) {
  163. };
  164. self.infoSyncTree =
  165. [[FSyncTree alloc] initWithListenProvider:infoListenProvider];
  166. FListenProvider *serverListenProvider = [[FListenProvider alloc] init];
  167. serverListenProvider.startListening =
  168. ^(FQuerySpec *query, NSNumber *tagId, id<FSyncTreeHash> hash,
  169. fbt_nsarray_nsstring onComplete) {
  170. [weakSelf.connection listen:query
  171. tagId:tagId
  172. hash:hash
  173. onComplete:^(NSString *status) {
  174. NSArray *events = onComplete(status);
  175. [weakSelf.eventRaiser raiseEvents:events];
  176. }];
  177. // No synchronous events for network-backed sync trees
  178. return @[];
  179. };
  180. serverListenProvider.stopListening = ^(FQuerySpec *query, NSNumber *tag) {
  181. [weakSelf.connection unlisten:query tagId:tag];
  182. };
  183. self.serverSyncTree =
  184. [[FSyncTree alloc] initWithPersistenceManager:self.persistenceManager
  185. listenProvider:serverListenProvider];
  186. [self restoreWrites];
  187. [self updateInfo:kDotInfoConnected withValue:@NO];
  188. [self setupNotifications];
  189. }
  190. - (void)restoreWrites {
  191. NSArray *writes = self.persistenceManager.userWrites;
  192. NSDictionary *serverValues =
  193. [FServerValues generateServerValues:self.serverClock];
  194. __block NSInteger lastWriteId = NSIntegerMin;
  195. [writes enumerateObjectsUsingBlock:^(FWriteRecord *write, NSUInteger idx,
  196. BOOL *stop) {
  197. NSInteger writeId = write.writeId;
  198. fbt_void_nsstring_nsstring callback =
  199. ^(NSString *status, NSString *errorReason) {
  200. [self warnIfWriteFailedAtPath:write.path
  201. status:status
  202. message:@"Persisted write"];
  203. [self ackWrite:writeId
  204. rerunTransactionsAtPath:write.path
  205. status:status];
  206. };
  207. if (lastWriteId >= writeId) {
  208. [NSException raise:NSInternalInconsistencyException
  209. format:@"Restored writes were not in order!"];
  210. }
  211. lastWriteId = writeId;
  212. self.writeIdCounter = writeId + 1;
  213. if ([write isOverwrite]) {
  214. FFLog(@"I-RDB038001", @"Restoring overwrite with id %ld",
  215. (long)write.writeId);
  216. [self.connection putData:[write.overwrite valForExport:YES]
  217. forPath:[write.path toString]
  218. withHash:nil
  219. withCallback:callback];
  220. id<FNode> resolved =
  221. [FServerValues resolveDeferredValueSnapshot:write.overwrite
  222. withSyncTree:self.serverSyncTree
  223. atPath:write.path
  224. serverValues:serverValues];
  225. [self.serverSyncTree applyUserOverwriteAtPath:write.path
  226. newData:resolved
  227. writeId:writeId
  228. isVisible:YES];
  229. } else {
  230. FFLog(@"I-RDB038002", @"Restoring merge with id %ld",
  231. (long)write.writeId);
  232. [self.connection mergeData:[write.merge valForExport:YES]
  233. forPath:[write.path toString]
  234. withCallback:callback];
  235. FCompoundWrite *resolved = [FServerValues
  236. resolveDeferredValueCompoundWrite:write.merge
  237. withSyncTree:self.serverSyncTree
  238. atPath:write.path
  239. serverValues:serverValues];
  240. [self.serverSyncTree applyUserMergeAtPath:write.path
  241. changedChildren:resolved
  242. writeId:writeId];
  243. }
  244. }];
  245. }
  246. - (NSString *)name {
  247. return self.repoInfo.namespace;
  248. }
  249. - (NSString *)description {
  250. return [self.repoInfo description];
  251. }
  252. - (void)interrupt {
  253. [self.connection interruptForReason:kFInterruptReasonRepoInterrupt];
  254. }
  255. - (void)resume {
  256. [self.connection resumeForReason:kFInterruptReasonRepoInterrupt];
  257. }
  258. // NOTE: Typically if you're calling this, you should be in an @autoreleasepool
  259. // block to make sure that ARC kicks in and cleans up things no longer
  260. // referenced (i.e. pendingPutsDB).
  261. - (void)dispose {
  262. [self.connection interruptForReason:kFInterruptReasonRepoInterrupt];
  263. // We need to nil out any references to LevelDB, to make sure the
  264. // LevelDB exclusive locks are released.
  265. [self.persistenceManager close];
  266. }
  267. - (NSInteger)nextWriteId {
  268. return self->_writeIdCounter++;
  269. }
  270. - (NSTimeInterval)serverTime {
  271. return [self.serverClock currentTime];
  272. }
  273. - (void)set:(FPath *)path
  274. withNode:(id<FNode>)node
  275. withCallback:(fbt_void_nserror_ref)onComplete {
  276. id value = [node valForExport:YES];
  277. FFLog(@"I-RDB038003", @"Setting: %@ with %@ pri: %@", [path toString],
  278. [value description], [[node getPriority] val]);
  279. // TODO: Optimize this behavior to either (a) store flag to skip resolving
  280. // where possible and / or (b) store unresolved paths on JSON parse
  281. NSDictionary *serverValues =
  282. [FServerValues generateServerValues:self.serverClock];
  283. id<FNode> existing = [self.serverSyncTree calcCompleteEventCacheAtPath:path
  284. excludeWriteIds:@[]];
  285. id<FNode> newNode =
  286. [FServerValues resolveDeferredValueSnapshot:node
  287. withExisting:existing
  288. serverValues:serverValues];
  289. NSInteger writeId = [self nextWriteId];
  290. [self.persistenceManager saveUserOverwrite:node
  291. atPath:path
  292. writeId:writeId];
  293. NSArray *events = [self.serverSyncTree applyUserOverwriteAtPath:path
  294. newData:newNode
  295. writeId:writeId
  296. isVisible:YES];
  297. [self.eventRaiser raiseEvents:events];
  298. [self.connection putData:value
  299. forPath:[path toString]
  300. withHash:nil
  301. withCallback:^(NSString *status, NSString *errorReason) {
  302. [self warnIfWriteFailedAtPath:path
  303. status:status
  304. message:@"setValue: or removeValue:"];
  305. [self ackWrite:writeId
  306. rerunTransactionsAtPath:path
  307. status:status];
  308. [self callOnComplete:onComplete
  309. withStatus:status
  310. errorReason:errorReason
  311. andPath:path];
  312. }];
  313. FPath *affectedPath = [self abortTransactionsAtPath:path
  314. error:kFTransactionSet];
  315. [self rerunTransactionsForPath:affectedPath];
  316. }
  317. - (void)update:(FPath *)path
  318. withNodes:(FCompoundWrite *)nodes
  319. withCallback:(fbt_void_nserror_ref)callback {
  320. NSDictionary *values = [nodes valForExport:YES];
  321. FFLog(@"I-RDB038004", @"Updating: %@ with %@", [path toString],
  322. [values description]);
  323. NSDictionary *serverValues =
  324. [FServerValues generateServerValues:self.serverClock];
  325. FCompoundWrite *resolved =
  326. [FServerValues resolveDeferredValueCompoundWrite:nodes
  327. withSyncTree:self.serverSyncTree
  328. atPath:path
  329. serverValues:serverValues];
  330. if (!resolved.isEmpty) {
  331. NSInteger writeId = [self nextWriteId];
  332. [self.persistenceManager saveUserMerge:nodes
  333. atPath:path
  334. writeId:writeId];
  335. NSArray *events = [self.serverSyncTree applyUserMergeAtPath:path
  336. changedChildren:resolved
  337. writeId:writeId];
  338. [self.eventRaiser raiseEvents:events];
  339. [self.connection mergeData:values
  340. forPath:[path description]
  341. withCallback:^(NSString *status, NSString *errorReason) {
  342. [self warnIfWriteFailedAtPath:path
  343. status:status
  344. message:@"updateChildValues:"];
  345. [self ackWrite:writeId
  346. rerunTransactionsAtPath:path
  347. status:status];
  348. [self callOnComplete:callback
  349. withStatus:status
  350. errorReason:errorReason
  351. andPath:path];
  352. }];
  353. [nodes enumerateWrites:^(FPath *childPath, id<FNode> node, BOOL *stop) {
  354. FPath *pathFromRoot = [path child:childPath];
  355. FFLog(@"I-RDB038005", @"Cancelling transactions at path: %@",
  356. pathFromRoot);
  357. FPath *affectedPath = [self abortTransactionsAtPath:pathFromRoot
  358. error:kFTransactionSet];
  359. [self rerunTransactionsForPath:affectedPath];
  360. }];
  361. } else {
  362. FFLog(@"I-RDB038006", @"update called with empty data. Doing nothing");
  363. // Do nothing, just call the callback
  364. [self callOnComplete:callback
  365. withStatus:@"ok"
  366. errorReason:nil
  367. andPath:path];
  368. }
  369. }
  370. - (void)onDisconnectCancel:(FPath *)path
  371. withCallback:(fbt_void_nserror_ref)callback {
  372. [self.connection
  373. onDisconnectCancelPath:path
  374. withCallback:^(NSString *status, NSString *errorReason) {
  375. BOOL success =
  376. [status isEqualToString:kFWPResponseForActionStatusOk];
  377. if (success) {
  378. [self.onDisconnect forgetPath:path];
  379. } else {
  380. FFLog(@"I-RDB038007",
  381. @"cancelDisconnectOperations: at %@ failed: %@",
  382. path, status);
  383. }
  384. [self callOnComplete:callback
  385. withStatus:status
  386. errorReason:errorReason
  387. andPath:path];
  388. }];
  389. }
  390. - (void)onDisconnectSet:(FPath *)path
  391. withNode:(id<FNode>)node
  392. withCallback:(fbt_void_nserror_ref)callback {
  393. [self.connection
  394. onDisconnectPutData:[node valForExport:YES]
  395. forPath:path
  396. withCallback:^(NSString *status, NSString *errorReason) {
  397. BOOL success =
  398. [status isEqualToString:kFWPResponseForActionStatusOk];
  399. if (success) {
  400. [self.onDisconnect rememberData:node onPath:path];
  401. } else {
  402. FFWarn(@"I-RDB038008",
  403. @"onDisconnectSetValue: or "
  404. @"onDisconnectRemoveValue: at %@ failed: %@",
  405. path, status);
  406. }
  407. [self callOnComplete:callback
  408. withStatus:status
  409. errorReason:errorReason
  410. andPath:path];
  411. }];
  412. }
  413. - (void)onDisconnectUpdate:(FPath *)path
  414. withNodes:(FCompoundWrite *)nodes
  415. withCallback:(fbt_void_nserror_ref)callback {
  416. if (!nodes.isEmpty) {
  417. NSDictionary *values = [nodes valForExport:YES];
  418. [self.connection
  419. onDisconnectMergeData:values
  420. forPath:path
  421. withCallback:^(NSString *status, NSString *errorReason) {
  422. BOOL success = [status
  423. isEqualToString:kFWPResponseForActionStatusOk];
  424. if (success) {
  425. [nodes enumerateWrites:^(FPath *relativePath,
  426. id<FNode> nodeUnresolved,
  427. BOOL *stop) {
  428. FPath *childPath = [path child:relativePath];
  429. [self.onDisconnect rememberData:nodeUnresolved
  430. onPath:childPath];
  431. }];
  432. } else {
  433. FFWarn(@"I-RDB038009",
  434. @"onDisconnectUpdateChildValues: at %@ "
  435. @"failed %@",
  436. path, status);
  437. }
  438. [self callOnComplete:callback
  439. withStatus:status
  440. errorReason:errorReason
  441. andPath:path];
  442. }];
  443. } else {
  444. // Do nothing, just call the callback
  445. [self callOnComplete:callback
  446. withStatus:@"ok"
  447. errorReason:nil
  448. andPath:path];
  449. }
  450. }
  451. - (void)purgeOutstandingWrites {
  452. FFLog(@"I-RDB038010", @"Purging outstanding writes");
  453. NSArray *events = [self.serverSyncTree removeAllWrites];
  454. [self.eventRaiser raiseEvents:events];
  455. // Abort any transactions
  456. [self abortTransactionsAtPath:[FPath empty] error:kFErrorWriteCanceled];
  457. // Remove outstanding writes from connection
  458. [self.connection purgeOutstandingWrites];
  459. }
  460. - (void)getData:(FIRDatabaseQuery *)query
  461. withCompletionBlock:
  462. (void (^_Nonnull)(NSError *__nullable error,
  463. FIRDataSnapshot *__nullable snapshot))block {
  464. FQuerySpec *querySpec = [query querySpec];
  465. [self.persistenceManager setQueryActive:querySpec];
  466. [self.connection
  467. getDataAtPath:[query.path toString]
  468. withParams:querySpec.params.wireProtocolParams
  469. withCallback:^(NSString *status, id data, NSString *errorReason) {
  470. id<FNode> node;
  471. if (![status isEqualToString:kFWPResponseForActionStatusOk]) {
  472. FFLog(@"I-RDB038024",
  473. @"getValue for query %@ falling back to cache",
  474. [querySpec.path toString]);
  475. node = [self.serverSyncTree
  476. calcCompleteEventCacheAtPath:querySpec.path
  477. excludeWriteIds:@[]];
  478. if ([node isEmpty]) {
  479. FFWarn(@"I-RDB038025",
  480. @"getValue for query at %@ failed: %@",
  481. [querySpec.path toString], status);
  482. NSDictionary *errorDict = @{
  483. NSLocalizedFailureReasonErrorKey : errorReason,
  484. NSLocalizedDescriptionKey : [NSString
  485. stringWithFormat:
  486. @"Unable to get latest value for query %@, "
  487. @"client offline and cache is empty",
  488. querySpec]
  489. };
  490. block([NSError errorWithDomain:kFirebaseCoreErrorDomain
  491. code:1
  492. userInfo:errorDict],
  493. nil);
  494. return;
  495. }
  496. } else {
  497. node = [FSnapshotUtilities nodeFrom:data];
  498. }
  499. [self.eventRaiser
  500. raiseEvents:[self.serverSyncTree
  501. applyServerOverwriteAtPath:[query path]
  502. newData:node]];
  503. block(nil,
  504. [[FIRDataSnapshot alloc]
  505. initWithRef:query.ref
  506. indexedNode:[FIndexedNode
  507. indexedNodeWithNode:node
  508. index:querySpec.index]]);
  509. [self.persistenceManager setQueryInactive:querySpec];
  510. }];
  511. }
  512. - (void)addEventRegistration:(id<FEventRegistration>)eventRegistration
  513. forQuery:(FQuerySpec *)query {
  514. NSArray *events = nil;
  515. if ([[query.path getFront] isEqualToString:kDotInfoPrefix]) {
  516. events = [self.infoSyncTree addEventRegistration:eventRegistration
  517. forQuery:query];
  518. } else {
  519. events = [self.serverSyncTree addEventRegistration:eventRegistration
  520. forQuery:query];
  521. }
  522. [self.eventRaiser raiseEvents:events];
  523. }
  524. - (void)removeEventRegistration:(id<FEventRegistration>)eventRegistration
  525. forQuery:(FQuerySpec *)query {
  526. // These are guaranteed not to raise events, since we're not passing in a
  527. // cancelError. However we can future-proof a little bit by handling the
  528. // return values anyways.
  529. FFLog(@"I-RDB038011", @"Removing event registration with hande: %lu",
  530. (unsigned long)eventRegistration.handle);
  531. NSArray *events = nil;
  532. if ([[query.path getFront] isEqualToString:kDotInfoPrefix]) {
  533. events = [self.infoSyncTree removeEventRegistration:eventRegistration
  534. forQuery:query
  535. cancelError:nil];
  536. } else {
  537. events = [self.serverSyncTree removeEventRegistration:eventRegistration
  538. forQuery:query
  539. cancelError:nil];
  540. }
  541. [self.eventRaiser raiseEvents:events];
  542. }
  543. - (void)keepQuery:(FQuerySpec *)query synced:(BOOL)synced {
  544. NSAssert(![[query.path getFront] isEqualToString:kDotInfoPrefix],
  545. @"Can't keep .info tree synced!");
  546. [self.serverSyncTree keepQuery:query synced:synced];
  547. }
  548. - (void)updateInfo:(NSString *)pathString withValue:(id)value {
  549. // hack to make serverTimeOffset available in a threadsafe way. Property is
  550. // marked as atomic
  551. if ([pathString isEqualToString:kDotInfoServerTimeOffset]) {
  552. NSTimeInterval offset = [(NSNumber *)value doubleValue] / 1000.0;
  553. self.serverClock =
  554. [[FOffsetClock alloc] initWithClock:[FSystemClock clock]
  555. offset:offset];
  556. }
  557. FPath *path = [[FPath alloc]
  558. initWith:[NSString
  559. stringWithFormat:@"%@/%@", kDotInfoPrefix, pathString]];
  560. id<FNode> newNode = [FSnapshotUtilities nodeFrom:value];
  561. [self.infoData updateSnapshot:path withNewSnapshot:newNode];
  562. NSArray *events = [self.infoSyncTree applyServerOverwriteAtPath:path
  563. newData:newNode];
  564. [self.eventRaiser raiseEvents:events];
  565. }
  566. - (void)callOnComplete:(fbt_void_nserror_ref)onComplete
  567. withStatus:(NSString *)status
  568. errorReason:(NSString *)errorReason
  569. andPath:(FPath *)path {
  570. if (onComplete) {
  571. FIRDatabaseReference *ref =
  572. [[FIRDatabaseReference alloc] initWithRepo:self path:path];
  573. BOOL statusOk = [status isEqualToString:kFWPResponseForActionStatusOk];
  574. NSError *err = nil;
  575. if (!statusOk) {
  576. err = [FUtilities errorForStatus:status andReason:errorReason];
  577. }
  578. [self.eventRaiser raiseCallback:^{
  579. onComplete(err, ref);
  580. }];
  581. }
  582. }
  583. - (void)ackWrite:(NSInteger)writeId
  584. rerunTransactionsAtPath:(FPath *)path
  585. status:(NSString *)status {
  586. if ([status isEqualToString:kFErrorWriteCanceled]) {
  587. // This write was already removed, we just need to ignore it...
  588. } else {
  589. BOOL success = [status isEqualToString:kFWPResponseForActionStatusOk];
  590. NSArray *clearEvents =
  591. [self.serverSyncTree ackUserWriteWithWriteId:writeId
  592. revert:!success
  593. persist:YES
  594. clock:self.serverClock];
  595. if ([clearEvents count] > 0) {
  596. [self rerunTransactionsForPath:path];
  597. }
  598. [self.eventRaiser raiseEvents:clearEvents];
  599. }
  600. }
  601. - (void)warnIfWriteFailedAtPath:(FPath *)path
  602. status:(NSString *)status
  603. message:(NSString *)message {
  604. if (!([status isEqualToString:kFWPResponseForActionStatusOk] ||
  605. [status isEqualToString:kFErrorWriteCanceled])) {
  606. FFWarn(@"I-RDB038012", @"%@ at %@ failed: %@", message, path, status);
  607. }
  608. }
  609. #pragma mark -
  610. #pragma mark FPersistentConnectionDelegate methods
  611. - (void)onDataUpdate:(FPersistentConnection *)fpconnection
  612. forPath:(NSString *)pathString
  613. message:(id)data
  614. isMerge:(BOOL)isMerge
  615. tagId:(NSNumber *)tagId {
  616. FFLog(@"I-RDB038013", @"onDataUpdateForPath: %@ withMessage: %@",
  617. pathString, data);
  618. // For testing.
  619. self.dataUpdateCount++;
  620. FPath *path = [[FPath alloc] initWith:pathString];
  621. data = self.interceptServerDataCallback
  622. ? self.interceptServerDataCallback(pathString, data)
  623. : data;
  624. NSArray *events = nil;
  625. if (tagId != nil) {
  626. if (isMerge) {
  627. NSDictionary *message = data;
  628. FCompoundWrite *taggedChildren =
  629. [FCompoundWrite compoundWriteWithValueDictionary:message];
  630. events =
  631. [self.serverSyncTree applyTaggedQueryMergeAtPath:path
  632. changedChildren:taggedChildren
  633. tagId:tagId];
  634. } else {
  635. id<FNode> taggedSnap = [FSnapshotUtilities nodeFrom:data];
  636. events =
  637. [self.serverSyncTree applyTaggedQueryOverwriteAtPath:path
  638. newData:taggedSnap
  639. tagId:tagId];
  640. }
  641. } else if (isMerge) {
  642. NSDictionary *message = data;
  643. FCompoundWrite *changedChildren =
  644. [FCompoundWrite compoundWriteWithValueDictionary:message];
  645. events = [self.serverSyncTree applyServerMergeAtPath:path
  646. changedChildren:changedChildren];
  647. } else {
  648. id<FNode> snap = [FSnapshotUtilities nodeFrom:data];
  649. events = [self.serverSyncTree applyServerOverwriteAtPath:path
  650. newData:snap];
  651. }
  652. if ([events count] > 0) {
  653. // Since we have a listener outstanding for each transaction, receiving
  654. // any events is a proxy for some change having occurred.
  655. [self rerunTransactionsForPath:path];
  656. }
  657. [self.eventRaiser raiseEvents:events];
  658. }
  659. - (void)onRangeMerge:(NSArray *)ranges
  660. forPath:(NSString *)pathString
  661. tagId:(NSNumber *)tag {
  662. FFLog(@"I-RDB038014", @"onRangeMerge: %@ => %@", pathString, ranges);
  663. // For testing
  664. self.rangeMergeUpdateCount++;
  665. FPath *path = [[FPath alloc] initWith:pathString];
  666. NSArray *events;
  667. if (tag != nil) {
  668. events = [self.serverSyncTree applyTaggedServerRangeMergeAtPath:path
  669. updates:ranges
  670. tagId:tag];
  671. } else {
  672. events = [self.serverSyncTree applyServerRangeMergeAtPath:path
  673. updates:ranges];
  674. }
  675. if (events.count > 0) {
  676. // Since we have a listener outstanding for each transaction, receiving
  677. // any events is a proxy for some change having occurred.
  678. [self rerunTransactionsForPath:path];
  679. }
  680. [self.eventRaiser raiseEvents:events];
  681. }
  682. - (void)onConnect:(FPersistentConnection *)fpconnection {
  683. [self updateInfo:kDotInfoConnected withValue:@YES];
  684. }
  685. - (void)onDisconnect:(FPersistentConnection *)fpconnection {
  686. [self updateInfo:kDotInfoConnected withValue:@NO];
  687. [self runOnDisconnectEvents];
  688. }
  689. - (void)onServerInfoUpdate:(FPersistentConnection *)fpconnection
  690. updates:(NSDictionary *)updates {
  691. for (NSString *key in updates) {
  692. id val = [updates objectForKey:key];
  693. [self updateInfo:key withValue:val];
  694. }
  695. }
  696. - (void)setupNotifications {
  697. NSString *const *backgroundConstant = (NSString *const *)dlsym(
  698. RTLD_DEFAULT, "UIApplicationDidEnterBackgroundNotification");
  699. if (backgroundConstant) {
  700. FFLog(@"I-RDB038015", @"Registering for background notification.");
  701. [[NSNotificationCenter defaultCenter]
  702. addObserver:self
  703. selector:@selector(didEnterBackground)
  704. name:*backgroundConstant
  705. object:nil];
  706. } else {
  707. FFLog(@"I-RDB038016",
  708. @"Skipped registering for background notification.");
  709. }
  710. }
  711. - (void)didEnterBackground {
  712. if (!self.config.persistenceEnabled)
  713. return;
  714. // Targetted compilation is ONLY for testing. UIKit is weak-linked in actual
  715. // release build.
  716. #if TARGET_OS_IOS || TARGET_OS_TV
  717. // The idea is to wait until any outstanding sets get written to disk. Since
  718. // the sets might still be in our dispatch queue, we wait for the dispatch
  719. // queue to catch up and for persistence to catch up. This may be
  720. // undesirable though. The dispatch queue might just be processing a bunch
  721. // of incoming data or something. We might want to keep track of whether
  722. // there are any unpersisted sets or something.
  723. FFLog(@"I-RDB038017",
  724. @"Entering background. Starting background task to finish work.");
  725. Class uiApplicationClass = NSClassFromString(@"UIApplication");
  726. assert(uiApplicationClass); // If we are here, we should be on iOS and
  727. // UIApplication should be available.
  728. UIApplication *application = [uiApplicationClass sharedApplication];
  729. __block UIBackgroundTaskIdentifier bgTask =
  730. [application beginBackgroundTaskWithExpirationHandler:^{
  731. [application endBackgroundTask:bgTask];
  732. }];
  733. NSDate *start = [NSDate date];
  734. dispatch_async([FIRDatabaseQuery sharedQueue], ^{
  735. NSTimeInterval finishTime = [start timeIntervalSinceNow] * -1;
  736. FFLog(@"I-RDB038018", @"Background task completed. Queue time: %f",
  737. finishTime);
  738. [application endBackgroundTask:bgTask];
  739. });
  740. #endif
  741. }
  742. #pragma mark -
  743. #pragma mark Internal methods
  744. /**
  745. * Applies all the changes stored up in the onDisconnect tree
  746. */
  747. - (void)runOnDisconnectEvents {
  748. FFLog(@"I-RDB038019", @"Running onDisconnectEvents");
  749. NSDictionary *serverValues =
  750. [FServerValues generateServerValues:self.serverClock];
  751. NSMutableArray *events = [[NSMutableArray alloc] init];
  752. [self.onDisconnect
  753. forEachTreeAtPath:[FPath empty]
  754. do:^(FPath *path, id<FNode> node) {
  755. id<FNode> existing = [self.serverSyncTree
  756. calcCompleteEventCacheAtPath:path
  757. excludeWriteIds:@[]];
  758. id<FNode> resolved = [FServerValues
  759. resolveDeferredValueSnapshot:node
  760. withExisting:existing
  761. serverValues:serverValues];
  762. [events addObjectsFromArray:
  763. [self.serverSyncTree
  764. applyServerOverwriteAtPath:path
  765. newData:resolved]];
  766. FPath *affectedPath =
  767. [self abortTransactionsAtPath:path
  768. error:kFTransactionSet];
  769. [self rerunTransactionsForPath:affectedPath];
  770. }];
  771. self.onDisconnect = [[FSparseSnapshotTree alloc] init];
  772. [self.eventRaiser raiseEvents:events];
  773. }
  774. - (NSDictionary *)dumpListens {
  775. return [self.connection dumpListens];
  776. }
  777. #pragma mark -
  778. #pragma mark Transactions
  779. /**
  780. * Setup the transaction data structures
  781. */
  782. - (void)initTransactions {
  783. self.transactionQueueTree = [[FTree alloc] init];
  784. self.hijackHash = NO;
  785. self.loggedTransactionPersistenceWarning = NO;
  786. }
  787. /**
  788. * Creates a new transaction, add its to the transactions we're tracking, and
  789. * sends it to the server if possible
  790. */
  791. - (void)startTransactionOnPath:(FPath *)path
  792. update:(fbt_transactionresult_mutabledata)update
  793. onComplete:(fbt_void_nserror_bool_datasnapshot)onComplete
  794. withLocalEvents:(BOOL)applyLocally {
  795. if (self.config.persistenceEnabled &&
  796. !self.loggedTransactionPersistenceWarning) {
  797. self.loggedTransactionPersistenceWarning = YES;
  798. FFInfo(@"I-RDB038020",
  799. @"runTransactionBlock: usage detected while persistence is "
  800. @"enabled. Please be aware that transactions "
  801. @"*will not* be persisted across app restarts. "
  802. @"See "
  803. @"https://www.firebase.com/docs/ios/guide/"
  804. @"offline-capabilities.html#section-handling-transactions-"
  805. @"offline for more details.");
  806. }
  807. FIRDatabaseReference *watchRef =
  808. [[FIRDatabaseReference alloc] initWithRepo:self path:path];
  809. // make sure we're listening on this node
  810. // Note: we can't do this asynchronously. To preserve event ordering, it has
  811. // to be done in this block. This is ok, this block is guaranteed to be our
  812. // own event loop
  813. NSUInteger handle = [[FUtilities LUIDGenerator] integerValue];
  814. fbt_void_datasnapshot cb = ^(FIRDataSnapshot *snapshot) {
  815. };
  816. FValueEventRegistration *registration =
  817. [[FValueEventRegistration alloc] initWithRepo:self
  818. handle:handle
  819. callback:cb
  820. cancelCallback:nil];
  821. [watchRef.repo addEventRegistration:registration
  822. forQuery:watchRef.querySpec];
  823. fbt_void_void unwatcher = ^{
  824. [watchRef removeObserverWithHandle:handle];
  825. };
  826. // Save all the data that represents this transaction
  827. FTupleTransaction *transaction = [[FTupleTransaction alloc] init];
  828. transaction.path = path;
  829. transaction.update = update;
  830. transaction.onComplete = onComplete;
  831. transaction.status = FTransactionInitializing;
  832. transaction.order = [FUtilities LUIDGenerator];
  833. transaction.applyLocally = applyLocally;
  834. transaction.retryCount = 0;
  835. transaction.unwatcher = unwatcher;
  836. transaction.currentWriteId = nil;
  837. transaction.currentInputSnapshot = nil;
  838. transaction.currentOutputSnapshotRaw = nil;
  839. transaction.currentOutputSnapshotResolved = nil;
  840. // Run transaction initially
  841. id<FNode> currentState = [self latestStateAtPath:path excludeWriteIds:nil];
  842. transaction.currentInputSnapshot = currentState;
  843. FIRMutableData *mutableCurrent =
  844. [[FIRMutableData alloc] initWithNode:currentState];
  845. FIRTransactionResult *result = transaction.update(mutableCurrent);
  846. if (!result.isSuccess) {
  847. // Abort the transaction
  848. transaction.unwatcher();
  849. transaction.currentOutputSnapshotRaw = nil;
  850. transaction.currentOutputSnapshotResolved = nil;
  851. if (transaction.onComplete) {
  852. FIRDatabaseReference *ref =
  853. [[FIRDatabaseReference alloc] initWithRepo:self
  854. path:transaction.path];
  855. FIndexedNode *indexedNode = [FIndexedNode
  856. indexedNodeWithNode:transaction.currentInputSnapshot];
  857. FIRDataSnapshot *snap =
  858. [[FIRDataSnapshot alloc] initWithRef:ref
  859. indexedNode:indexedNode];
  860. [self.eventRaiser raiseCallback:^{
  861. transaction.onComplete(nil, NO, snap);
  862. }];
  863. }
  864. } else {
  865. // Note: different from js. We don't need to validate, FIRMutableData
  866. // does validation. We also don't have to worry about priorities. Just
  867. // mark as run and add to queue.
  868. transaction.status = FTransactionRun;
  869. FTree *queueNode = [self.transactionQueueTree subTree:transaction.path];
  870. NSMutableArray *nodeQueue = [queueNode getValue];
  871. if (nodeQueue == nil) {
  872. nodeQueue = [[NSMutableArray alloc] init];
  873. }
  874. [nodeQueue addObject:transaction];
  875. [queueNode setValue:nodeQueue];
  876. // Update visibleData and raise events
  877. // Note: We intentionally raise events after updating all of our
  878. // transaction state, since the user could start new transactions from
  879. // the event callbacks
  880. NSDictionary *serverValues =
  881. [FServerValues generateServerValues:self.serverClock];
  882. id<FNode> newValUnresolved = [result.update nodeValue];
  883. id<FNode> newVal =
  884. [FServerValues resolveDeferredValueSnapshot:newValUnresolved
  885. withExisting:currentState
  886. serverValues:serverValues];
  887. transaction.currentOutputSnapshotRaw = newValUnresolved;
  888. transaction.currentOutputSnapshotResolved = newVal;
  889. transaction.currentWriteId =
  890. [NSNumber numberWithInteger:[self nextWriteId]];
  891. NSArray *events = [self.serverSyncTree
  892. applyUserOverwriteAtPath:path
  893. newData:newVal
  894. writeId:[transaction.currentWriteId integerValue]
  895. isVisible:transaction.applyLocally];
  896. [self.eventRaiser raiseEvents:events];
  897. [self sendAllReadyTransactions];
  898. }
  899. }
  900. /**
  901. * @param writeIdsToExclude A specific set to exclude
  902. */
  903. - (id<FNode>)latestStateAtPath:(FPath *)path
  904. excludeWriteIds:(NSArray *)writeIdsToExclude {
  905. id<FNode> latestState =
  906. [self.serverSyncTree calcCompleteEventCacheAtPath:path
  907. excludeWriteIds:writeIdsToExclude];
  908. return latestState ? latestState : [FEmptyNode emptyNode];
  909. }
  910. /**
  911. * Sends any already-run transactions that aren't waiting for outstanding
  912. * transactions to complete.
  913. *
  914. * Externally, call the version with no arguments.
  915. * Internally, calls itself recursively with a particular transactionQueueTree
  916. * node to recurse through the tree
  917. */
  918. - (void)sendAllReadyTransactions {
  919. FTree *node = self.transactionQueueTree;
  920. [self pruneCompletedTransactionsBelowNode:node];
  921. [self sendReadyTransactionsForTree:node];
  922. }
  923. - (void)sendReadyTransactionsForTree:(FTree *)node {
  924. NSMutableArray *queue = [node getValue];
  925. if (queue != nil) {
  926. queue = [self buildTransactionQueueAtNode:node];
  927. NSAssert([queue count] > 0, @"Sending zero length transaction queue");
  928. NSUInteger notRunIndex = [queue
  929. indexOfObjectPassingTest:^BOOL(id obj, NSUInteger idx, BOOL *stop) {
  930. return ((FTupleTransaction *)obj).status != FTransactionRun;
  931. }];
  932. // If they're all run (and not sent), we can send them. Else, we must
  933. // wait.
  934. if (notRunIndex == NSNotFound) {
  935. [self sendTransactionQueue:queue atPath:node.path];
  936. }
  937. } else if ([node hasChildren]) {
  938. [node forEachChild:^(FTree *child) {
  939. [self sendReadyTransactionsForTree:child];
  940. }];
  941. }
  942. }
  943. /**
  944. * Given a list of run transactions, send them to the server and then handle the
  945. * result (success or failure).
  946. */
  947. - (void)sendTransactionQueue:(NSMutableArray *)queue atPath:(FPath *)path {
  948. // Mark transactions as sent and bump the retry count
  949. NSMutableArray *writeIdsToExclude = [[NSMutableArray alloc] init];
  950. for (FTupleTransaction *transaction in queue) {
  951. [writeIdsToExclude addObject:transaction.currentWriteId];
  952. }
  953. id<FNode> latestState = [self latestStateAtPath:path
  954. excludeWriteIds:writeIdsToExclude];
  955. id<FNode> snapToSend = latestState;
  956. NSString *latestHash = [latestState dataHash];
  957. for (FTupleTransaction *transaction in queue) {
  958. NSAssert(
  959. transaction.status == FTransactionRun,
  960. @"[FRepo sendTransactionQueue:] items in queue should all be run.");
  961. FFLog(@"I-RDB038021", @"Transaction at %@ set to SENT",
  962. transaction.path);
  963. transaction.status = FTransactionSent;
  964. transaction.retryCount++;
  965. FPath *relativePath = [FPath relativePathFrom:path to:transaction.path];
  966. // If we've gotten to this point, the output snapshot must be defined.
  967. snapToSend =
  968. [snapToSend updateChild:relativePath
  969. withNewChild:transaction.currentOutputSnapshotRaw];
  970. }
  971. id dataToSend = [snapToSend valForExport:YES];
  972. NSString *pathToSend = [path description];
  973. latestHash = self.hijackHash ? @"badhash" : latestHash;
  974. // Send the put
  975. [self.connection
  976. putData:dataToSend
  977. forPath:pathToSend
  978. withHash:latestHash
  979. withCallback:^(NSString *status, NSString *errorReason) {
  980. FFLog(@"I-RDB038022", @"Transaction put response: %@ : %@",
  981. pathToSend, status);
  982. NSMutableArray *events = [[NSMutableArray alloc] init];
  983. if ([status isEqualToString:kFWPResponseForActionStatusOk]) {
  984. // Queue up the callbacks and fire them after cleaning up all of
  985. // our transaction state, since the callback could trigger more
  986. // transactions or sets.
  987. NSMutableArray *callbacks = [[NSMutableArray alloc] init];
  988. for (FTupleTransaction *transaction in queue) {
  989. transaction.status = FTransactionCompleted;
  990. [events addObjectsFromArray:
  991. [self.serverSyncTree
  992. ackUserWriteWithWriteId:
  993. [transaction.currentWriteId integerValue]
  994. revert:NO
  995. persist:NO
  996. clock:self.serverClock]];
  997. if (transaction.onComplete) {
  998. // We never unset the output snapshot, and given that this
  999. // transaction is complete, it should be set
  1000. id<FNode> node =
  1001. transaction.currentOutputSnapshotResolved;
  1002. FIndexedNode *indexedNode =
  1003. [FIndexedNode indexedNodeWithNode:node];
  1004. FIRDatabaseReference *ref = [[FIRDatabaseReference alloc]
  1005. initWithRepo:self
  1006. path:transaction.path];
  1007. FIRDataSnapshot *snapshot =
  1008. [[FIRDataSnapshot alloc] initWithRef:ref
  1009. indexedNode:indexedNode];
  1010. fbt_void_void cb = ^{
  1011. transaction.onComplete(nil, YES, snapshot);
  1012. };
  1013. [callbacks addObject:[cb copy]];
  1014. }
  1015. transaction.unwatcher();
  1016. }
  1017. // Now remove the completed transactions.
  1018. [self
  1019. pruneCompletedTransactionsBelowNode:[self.transactionQueueTree
  1020. subTree:path]];
  1021. // There may be pending transactions that we can now send.
  1022. [self sendAllReadyTransactions];
  1023. // Finally, trigger onComplete callbacks
  1024. [self.eventRaiser raiseCallbacks:callbacks];
  1025. } else {
  1026. // transactions are no longer sent. Update their status
  1027. // appropriately.
  1028. if ([status
  1029. isEqualToString:kFWPResponseForActionStatusDataStale]) {
  1030. for (FTupleTransaction *transaction in queue) {
  1031. if (transaction.status == FTransactionSentNeedsAbort) {
  1032. transaction.status = FTransactionNeedsAbort;
  1033. } else {
  1034. transaction.status = FTransactionRun;
  1035. }
  1036. }
  1037. } else {
  1038. FFWarn(@"I-RDB038023",
  1039. @"runTransactionBlock: at %@ failed: %@", path,
  1040. status);
  1041. for (FTupleTransaction *transaction in queue) {
  1042. transaction.status = FTransactionNeedsAbort;
  1043. [transaction setAbortStatus:status reason:errorReason];
  1044. }
  1045. }
  1046. }
  1047. [self rerunTransactionsForPath:path];
  1048. [self.eventRaiser raiseEvents:events];
  1049. }];
  1050. }
  1051. /**
  1052. * Finds all transactions dependent on the data at changed Path and reruns them.
  1053. *
  1054. * Should be called any time cached data changes.
  1055. *
  1056. * Return the highest path that was affected by rerunning transactions. This is
  1057. * the path at which events need to be raised for.
  1058. */
  1059. - (FPath *)rerunTransactionsForPath:(FPath *)changedPath {
  1060. // For the common case that there are no transactions going on, skip all
  1061. // this!
  1062. if ([self.transactionQueueTree isEmpty]) {
  1063. return changedPath;
  1064. } else {
  1065. FTree *rootMostTransactionNode =
  1066. [self getAncestorTransactionNodeForPath:changedPath];
  1067. FPath *path = rootMostTransactionNode.path;
  1068. NSArray *queue =
  1069. [self buildTransactionQueueAtNode:rootMostTransactionNode];
  1070. [self rerunTransactionQueue:queue atPath:path];
  1071. return path;
  1072. }
  1073. }
  1074. /**
  1075. * Does all the work of rerunning transactions (as well as cleans up aborted
  1076. * transactions and whatnot).
  1077. */
  1078. - (void)rerunTransactionQueue:(NSArray *)queue atPath:(FPath *)path {
  1079. if (queue.count == 0) {
  1080. return; // nothing to do
  1081. }
  1082. // Queue up the callbacks and fire them after cleaning up all of our
  1083. // transaction state, since the callback could trigger more transactions or
  1084. // sets.
  1085. NSMutableArray *events = [[NSMutableArray alloc] init];
  1086. NSMutableArray *callbacks = [[NSMutableArray alloc] init];
  1087. // Ignore, by default, all of the sets in this queue, since we're re-running
  1088. // all of them. However, we want to include the results of new sets
  1089. // triggered as part of this re-run, so we don't want to ignore a range,
  1090. // just these specific sets.
  1091. NSMutableArray *writeIdsToExclude = [[NSMutableArray alloc] init];
  1092. for (FTupleTransaction *transaction in queue) {
  1093. [writeIdsToExclude addObject:transaction.currentWriteId];
  1094. }
  1095. for (FTupleTransaction *transaction in queue) {
  1096. FPath *relativePath __unused =
  1097. [FPath relativePathFrom:path to:transaction.path];
  1098. BOOL abortTransaction = NO;
  1099. NSAssert(relativePath != nil, @"[FRepo rerunTransactionsQueue:] "
  1100. @"relativePath should not be null.");
  1101. if (transaction.status == FTransactionNeedsAbort) {
  1102. abortTransaction = YES;
  1103. if (![transaction.abortStatus
  1104. isEqualToString:kFErrorWriteCanceled]) {
  1105. NSArray *ackEvents = [self.serverSyncTree
  1106. ackUserWriteWithWriteId:[transaction.currentWriteId
  1107. integerValue]
  1108. revert:YES
  1109. persist:NO
  1110. clock:self.serverClock];
  1111. [events addObjectsFromArray:ackEvents];
  1112. }
  1113. } else if (transaction.status == FTransactionRun) {
  1114. if (transaction.retryCount >= kFTransactionMaxRetries) {
  1115. abortTransaction = YES;
  1116. [transaction setAbortStatus:kFTransactionTooManyRetries
  1117. reason:nil];
  1118. [events
  1119. addObjectsFromArray:
  1120. [self.serverSyncTree
  1121. ackUserWriteWithWriteId:[transaction.currentWriteId
  1122. integerValue]
  1123. revert:YES
  1124. persist:NO
  1125. clock:self.serverClock]];
  1126. } else {
  1127. // This code reruns a transaction
  1128. id<FNode> currentNode =
  1129. [self latestStateAtPath:transaction.path
  1130. excludeWriteIds:writeIdsToExclude];
  1131. transaction.currentInputSnapshot = currentNode;
  1132. FIRMutableData *mutableCurrent =
  1133. [[FIRMutableData alloc] initWithNode:currentNode];
  1134. FIRTransactionResult *result =
  1135. transaction.update(mutableCurrent);
  1136. if (result.isSuccess) {
  1137. NSNumber *oldWriteId = transaction.currentWriteId;
  1138. NSDictionary *serverValues =
  1139. [FServerValues generateServerValues:self.serverClock];
  1140. id<FNode> newVal = [result.update nodeValue];
  1141. id<FNode> newValResolved = [FServerValues
  1142. resolveDeferredValueSnapshot:newVal
  1143. withExisting:transaction
  1144. .currentInputSnapshot
  1145. serverValues:serverValues];
  1146. transaction.currentOutputSnapshotRaw = newVal;
  1147. transaction.currentOutputSnapshotResolved = newValResolved;
  1148. transaction.currentWriteId =
  1149. [NSNumber numberWithInteger:[self nextWriteId]];
  1150. // Mutates writeIdsToExclude in place
  1151. [writeIdsToExclude removeObject:oldWriteId];
  1152. [events
  1153. addObjectsFromArray:
  1154. [self.serverSyncTree
  1155. applyUserOverwriteAtPath:transaction.path
  1156. newData:
  1157. transaction
  1158. .currentOutputSnapshotResolved
  1159. writeId:
  1160. [transaction.currentWriteId
  1161. integerValue]
  1162. isVisible:transaction
  1163. .applyLocally]];
  1164. [events addObjectsFromArray:
  1165. [self.serverSyncTree
  1166. ackUserWriteWithWriteId:[oldWriteId
  1167. integerValue]
  1168. revert:YES
  1169. persist:NO
  1170. clock:self.serverClock]];
  1171. } else {
  1172. abortTransaction = YES;
  1173. // The user aborted the transaction. JS treats ths as a
  1174. // "nodata" abort, but it's not an error, so we don't send
  1175. // them an error.
  1176. [transaction setAbortStatus:nil reason:nil];
  1177. [events
  1178. addObjectsFromArray:
  1179. [self.serverSyncTree
  1180. ackUserWriteWithWriteId:
  1181. [transaction.currentWriteId integerValue]
  1182. revert:YES
  1183. persist:NO
  1184. clock:self.serverClock]];
  1185. }
  1186. }
  1187. }
  1188. [self.eventRaiser raiseEvents:events];
  1189. events = nil;
  1190. if (abortTransaction) {
  1191. // Abort
  1192. transaction.status = FTransactionCompleted;
  1193. transaction.unwatcher();
  1194. if (transaction.onComplete) {
  1195. FIRDatabaseReference *ref = [[FIRDatabaseReference alloc]
  1196. initWithRepo:self
  1197. path:transaction.path];
  1198. FIndexedNode *lastInput = [FIndexedNode
  1199. indexedNodeWithNode:transaction.currentInputSnapshot];
  1200. FIRDataSnapshot *snap =
  1201. [[FIRDataSnapshot alloc] initWithRef:ref
  1202. indexedNode:lastInput];
  1203. fbt_void_void cb = ^{
  1204. // Unlike JS, no need to check for "nodata" because ObjC has
  1205. // abortError = nil
  1206. transaction.onComplete(transaction.abortError, NO, snap);
  1207. };
  1208. [callbacks addObject:[cb copy]];
  1209. }
  1210. }
  1211. }
  1212. // Note: unlike current js client, we don't need to preserve priority. Users
  1213. // can set priority via FIRMutableData
  1214. // Clean up completed transactions.
  1215. [self pruneCompletedTransactionsBelowNode:self.transactionQueueTree];
  1216. // Now fire callbacks, now that we're in a good, known state.
  1217. [self.eventRaiser raiseCallbacks:callbacks];
  1218. // Try to send the transaction result to the server
  1219. [self sendAllReadyTransactions];
  1220. }
  1221. - (FTree *)getAncestorTransactionNodeForPath:(FPath *)path {
  1222. FTree *transactionNode = self.transactionQueueTree;
  1223. while (![path isEmpty] && [transactionNode getValue] == nil) {
  1224. NSString *front = [path getFront];
  1225. transactionNode =
  1226. [transactionNode subTree:[[FPath alloc] initWith:front]];
  1227. path = [path popFront];
  1228. }
  1229. return transactionNode;
  1230. }
  1231. - (NSMutableArray *)buildTransactionQueueAtNode:(FTree *)node {
  1232. NSMutableArray *queue = [[NSMutableArray alloc] init];
  1233. [self aggregateTransactionQueuesForNode:node andQueue:queue];
  1234. [queue sortUsingComparator:^NSComparisonResult(FTupleTransaction *obj1,
  1235. FTupleTransaction *obj2) {
  1236. return [obj1.order compare:obj2.order];
  1237. }];
  1238. return queue;
  1239. }
  1240. - (void)aggregateTransactionQueuesForNode:(FTree *)node
  1241. andQueue:(NSMutableArray *)queue {
  1242. NSArray *nodeQueue = [node getValue];
  1243. [queue addObjectsFromArray:nodeQueue];
  1244. [node forEachChild:^(FTree *child) {
  1245. [self aggregateTransactionQueuesForNode:child andQueue:queue];
  1246. }];
  1247. }
  1248. /**
  1249. * Remove COMPLETED transactions at or below this node in the
  1250. * transactionQueueTree
  1251. */
  1252. - (void)pruneCompletedTransactionsBelowNode:(FTree *)node {
  1253. NSMutableArray *queue = [node getValue];
  1254. if (queue != nil) {
  1255. int i = 0;
  1256. // remove all of the completed transactions from the queue
  1257. while (i < queue.count) {
  1258. FTupleTransaction *transaction = [queue objectAtIndex:i];
  1259. if (transaction.status == FTransactionCompleted) {
  1260. [queue removeObjectAtIndex:i];
  1261. } else {
  1262. i++;
  1263. }
  1264. }
  1265. if (queue.count > 0) {
  1266. [node setValue:queue];
  1267. } else {
  1268. [node setValue:nil];
  1269. }
  1270. }
  1271. [node forEachChildMutationSafe:^(FTree *child) {
  1272. [self pruneCompletedTransactionsBelowNode:child];
  1273. }];
  1274. }
  1275. /**
  1276. * Aborts all transactions on ancestors or descendants of the specified path.
  1277. * Called when doing a setValue: or updateChildValues: since we consider them
  1278. * incompatible with transactions
  1279. *
  1280. * @param path path for which we want to abort related transactions.
  1281. */
  1282. - (FPath *)abortTransactionsAtPath:(FPath *)path error:(NSString *)error {
  1283. // For the common case that there are no transactions going on, skip all
  1284. // this!
  1285. if ([self.transactionQueueTree isEmpty]) {
  1286. return path;
  1287. } else {
  1288. FPath *affectedPath =
  1289. [self getAncestorTransactionNodeForPath:path].path;
  1290. FTree *transactionNode = [self.transactionQueueTree subTree:path];
  1291. [transactionNode forEachAncestor:^BOOL(FTree *ancestor) {
  1292. [self abortTransactionsAtNode:ancestor error:error];
  1293. return NO;
  1294. }];
  1295. [self abortTransactionsAtNode:transactionNode error:error];
  1296. [transactionNode forEachDescendant:^(FTree *child) {
  1297. [self abortTransactionsAtNode:child error:error];
  1298. }];
  1299. return affectedPath;
  1300. }
  1301. }
  1302. /**
  1303. * Abort transactions stored in this transactions queue node.
  1304. *
  1305. * @param node Node to abort transactions for.
  1306. */
  1307. - (void)abortTransactionsAtNode:(FTree *)node error:(NSString *)error {
  1308. NSMutableArray *queue = [node getValue];
  1309. if (queue != nil) {
  1310. // Queue up the callbacks and fire them after cleaning up all of our
  1311. // transaction state, since can be immediately aborted and removed.
  1312. NSMutableArray *callbacks = [[NSMutableArray alloc] init];
  1313. // Go through queue. Any already-sent transactions must be marked for
  1314. // abort, while the unsent ones can be immediately aborted and removed
  1315. NSMutableArray *events = [[NSMutableArray alloc] init];
  1316. int lastSent = -1;
  1317. // Note: all of the sent transactions will be at the front of the queue,
  1318. // so safe to increment lastSent
  1319. for (FTupleTransaction *transaction in queue) {
  1320. if (transaction.status == FTransactionSentNeedsAbort) {
  1321. // No-op. already marked.
  1322. } else if (transaction.status == FTransactionSent) {
  1323. // Mark this transaction for abort when it returns
  1324. lastSent++;
  1325. transaction.status = FTransactionSentNeedsAbort;
  1326. [transaction setAbortStatus:error reason:nil];
  1327. } else {
  1328. // we can abort this immediately
  1329. transaction.unwatcher();
  1330. if ([error isEqualToString:kFTransactionSet]) {
  1331. [events
  1332. addObjectsFromArray:
  1333. [self.serverSyncTree
  1334. ackUserWriteWithWriteId:
  1335. [transaction.currentWriteId integerValue]
  1336. revert:YES
  1337. persist:NO
  1338. clock:self.serverClock]];
  1339. } else {
  1340. // If it was cancelled it was already removed from the sync
  1341. // tree, no need to ack
  1342. NSAssert([error isEqualToString:kFErrorWriteCanceled], nil);
  1343. }
  1344. if (transaction.onComplete) {
  1345. NSError *abortReason = [FUtilities errorForStatus:error
  1346. andReason:nil];
  1347. FIRDataSnapshot *snapshot = nil;
  1348. fbt_void_void cb = ^{
  1349. transaction.onComplete(abortReason, NO, snapshot);
  1350. };
  1351. [callbacks addObject:[cb copy]];
  1352. }
  1353. }
  1354. }
  1355. if (lastSent == -1) {
  1356. // We're not waiting for any sent transactions. We can clear the
  1357. // queue.
  1358. [node setValue:nil];
  1359. } else {
  1360. // Remove the transactions we aborted
  1361. NSRange theRange;
  1362. theRange.location = lastSent + 1;
  1363. theRange.length = queue.count - theRange.location;
  1364. [queue removeObjectsInRange:theRange];
  1365. }
  1366. // Now fire the callbacks
  1367. [self.eventRaiser raiseEvents:events];
  1368. [self.eventRaiser raiseCallbacks:callbacks];
  1369. }
  1370. }
  1371. @end