@ -10,6 +10,12 @@ import DifferenceKit
// /
// / * * N o t e : * * W e * * M U S T * * h a v e a c c u r a t e ` f i l t e r S Q L ` a n d ` o r d e r S Q L ` v a l u e s o t h e r w i s e t h e i n d e x i n g w o n ' t w o r k
public class PagedDatabaseObserver < ObservedTable , T > : TransactionObserver where ObservedTable : TableRecord & ColumnExpressible & Identifiable , T : FetchableRecordWithRowId & Identifiable {
private let commitProcessingQueue : DispatchQueue = DispatchQueue (
label : " PagedDatabaseObserver.commitProcessingQueue " ,
qos : . userInitiated ,
attributes : [ ] // M u s t b e s e r i a l i n o r d e r t o a v o i d u p d a t e s g e t t i n g p r o c e s s e d i n t h e w r o n g o r d e r
)
// MARK: - V a r i a b l e s
private let pagedTableName : String
@ -145,74 +151,58 @@ public class PagedDatabaseObserver<ObservedTable, T>: TransactionObserver where
changesInCommit . mutate { $0 . insert ( trackedChange ) }
}
// N o t e : W e w i l l p r o c e s s a l l u p d a t e s w h i c h c o m e t h r o u g h t h i s m e t h o d e v e n i f
// ' o n C h a n g e ' i s n u l l b e c a u s e i f t h e U I s t o p s o b s e r v i n g a n d t h e n s t a r t s a g a i n
// l a t e r w e d o n ' t w a n t t o h a v e m i s s e d a n y c h a n g e s w h i c h h a p p e n e d w h i l e t h e U I
// w a s n ' t s u b s c r i b e d ( a n d d o i n g a f u l l r e - q u e r y s e e m s p a i n f u l . . . )
// / W e w i l l p r o c e s s a l l u p d a t e s w h i c h c o m e t h r o u g h t h i s m e t h o d e v e n i f ' o n C h a n g e ' i s n u l l b e c a u s e i f t h e U I s t o p s o b s e r v i n g a n d t h e n s t a r t s
// / a g a i n l a t e r w e d o n ' t w a n t t o h a v e m i s s e d a n y c h a n g e s w h i c h h a p p e n e d w h i l e t h e U I w a s n ' t s u b s c r i b e d ( a n d d o i n g a f u l l r e - q u e r y s e e m s p a i n f u l . . . )
// /
// / * * N o t e : * * T h i s f u n c t i o n i s g e n e r a l l y c a l l e d w i t h i n t h e D B W r i t e t h r e a d b u t w e d o n ' t a c t u a l l y n e e d w r i t e a c c e s s t o p r o c e s s t h e c o m m i t , i n o r d e r
// / t o a v o i d b l o c k i n g t h e D B W r i t e t h r e a d w e d i s p a t c h t o a s e r i a l ` c o m m i t P r o c e s s i n g Q u e u e ` t o p r o c e s s t h e i n c o m i n g c h a n g e s ( i n t h e p a s t n o t d o i n g
// / s o w a s r e s u l t i n g i n h a n g i n g w h e n t h e r e w a s a l o t o f a c t i v i t y h a p p e n i n g )
public func databaseDidCommit ( _ db : Database ) {
// S i n c e w e c a n ' t b e s u r e t h e b e h a v i o u r s o f ' d a t a b a s e D i d C h a n g e ' a n d ' d a t a b a s e D i d C o m m i t ' w o n ' t c h a n g e i n
// t h e f u t u r e w e e x t r a c t a n d c l e a r t h e v a l u e s i n ' c h a n g e s I n C o m m i t ' s i n c e i t ' s ' A t o m i c < T > ' s o w i l l d i f f e r e n t
// t h r e a d s m o d i f y i n g t h e d a t a r e s u l t i n g i n u s m i s s i n g a c h a n g e
var committedChanges : Set < PagedData . TrackedChange > = [ ]
self . changesInCommit . mutate { cachedChanges in
committedChanges = cachedChanges
cachedChanges . removeAll ( )
}
// N o t e : T h i s m e t h o d w i l l b e c a l l e d r e g a r d l e s s o f w h e t h e r t h e r e w e r e a c t u a l l y c h a n g e s
// i n t h e a r e a s w e a r e o b s e r v i n g s o w e w a n t t o e a r l y - o u t i f t h e r e a r e n ' t a n y r e l e v a n t
// u p d a t e d r o w s
commitProcessingQueue . async { [ weak self ] in
self ? . processDatabaseCommit ( committedChanges : committedChanges )
}
}
private func processDatabaseCommit ( committedChanges : Set < PagedData . TrackedChange > ) {
// D o n o t h i n g w h e n t h e r e a r e n o c h a n g e s
guard ! committedChanges . isEmpty else { return }
typealias AssociatedDataInfo = [ ( hasChanges : Bool , data : ErasedAssociatedRecord ) ]
typealias UpdatedData = ( cache : DataCache < T > , pageInfo : PagedData . PageInfo , hasChanges : Bool , associatedData : AssociatedDataInfo )
// S t o r e t h e i n s t a n c e v a r i a b l e s l o c a l l y t o a v o i d u n w r a p p i n g
let dataCache : DataCache < T > = self . dataCache . wrappedValue
let pageInfo : PagedData . PageInfo = self . pageInfo . wrappedValue
let joinSQL : SQL ? = self . joinSQL
let orderSQL : SQL = self . orderSQL
let filterSQL : SQL = self . filterSQL
let associatedRecords : [ ErasedAssociatedRecord ] = self . associatedRecords
let updateDataAndCallbackIfNeeded : ( DataCache < T > , PagedData . PageInfo , Bool ) -> ( ) = { [ weak self ] updatedDataCache , updatedPageInfo , cacheHasChanges in
let associatedDataInfo : [ ( hasChanges : Bool , data : ErasedAssociatedRecord ) ] = associatedRecords
. map { associatedRecord in
let hasChanges : Bool = associatedRecord . tryUpdateForDatabaseCommit (
db ,
changes : committedChanges ,
joinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL ,
pageInfo : updatedPageInfo
)
return ( hasChanges , associatedRecord )
}
// C h e c k i f w e n e e d t o t r i g g e r a c h a n g e c a l l b a c k
guard cacheHasChanges || associatedDataInfo . contains ( where : { hasChanges , _ in hasChanges } ) else {
return
}
// I f t h e a s s o c i a t e d d a t a c h a n g e d t h e n u p d a t e t h e u p d a t e d C a c h e d D a t a w i t h t h e
// u p d a t e d a s s o c i a t e d d a t a
var finalUpdatedDataCache : DataCache < T > = updatedDataCache
associatedDataInfo . forEach { hasChanges , associatedData in
guard cacheHasChanges || hasChanges else { return }
let getAssociatedDataInfo : ( Database , PagedData . PageInfo ) -> AssociatedDataInfo = { db , updatedPageInfo in
associatedRecords . map { associatedRecord in
let hasChanges : Bool = associatedRecord . tryUpdateForDatabaseCommit (
db ,
changes : committedChanges ,
joinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL ,
pageInfo : updatedPageInfo
)
finalUpdatedDataCache = associatedData . updateAssociatedData ( to : finalUpdatedDataCache )
return ( hasChanges , associatedRecord )
}
// U p d a t e t h e c a c h e , p a g e I n f o a n d t h e c h a n g e c a l l b a c k
self ? . dataCache . mutate { $0 = finalUpdatedDataCache }
self ? . pageInfo . mutate { $0 = updatedPageInfo }
// M a k e s u r e t h e u p d a t e s r u n o n t h e m a i n t h r e a d
guard Thread . isMainThread else {
DispatchQueue . main . async { [ weak self ] in
self ? . onChangeUnsorted ( finalUpdatedDataCache . values , updatedPageInfo )
}
return
}
self ? . onChangeUnsorted ( finalUpdatedDataCache . values , updatedPageInfo )
}
// D e t e r m i n g i f t h e r e w e r e a n y d i r e c t o r r e l a t e d d a t a c h a n g e s
// D e t e r m i n e i f t h e r e w e r e a n y d i r e c t o r r e l a t e d d a t a c h a n g e s
let directChanges : Set < PagedData . TrackedChange > = committedChanges
. filter { $0 . tableName = = pagedTableName }
let relatedChanges : [ String : [ PagedData . TrackedChange ] ] = committedChanges
@ -227,215 +217,248 @@ public class PagedDatabaseObserver<ObservedTable, T>: TransactionObserver where
. filter { $0 . tableName != pagedTableName }
. filter { $0 . kind = = . delete }
guard ! directChanges . isEmpty || ! relatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
updateDataAndCallbackIfNeeded ( self . dataCache . wrappedValue , self . pageInfo . wrappedValue , false )
return
}
var updatedPageInfo : PagedData . PageInfo = self . pageInfo . wrappedValue
var updatedDataCache : DataCache < T > = self . dataCache . wrappedValue
let deletionChanges : [ Int64 ] = directChanges
. filter { $0 . kind = = . delete }
. map { $0 . rowId }
let oldDataCount : Int = dataCache . wrappedValue . count
// F i r s t r e m o v e a n y i t e m s w h i c h h a v e b e e n d e l e t e d
if ! deletionChanges . isEmpty {
updatedDataCache = updatedDataCache . deleting ( rowIds : deletionChanges )
// M a k e s u r e t h e r e w e r e a c t u a l l y c h a n g e s
if updatedDataCache . count != oldDataCount {
let dataSizeDiff : Int = ( updatedDataCache . count - oldDataCount )
// P r o c e s s a n d r e t r i e v e t h e u p d a t e d d a t a
let updatedData : UpdatedData = Storage . shared
. read { db -> UpdatedData in
// I f t h e r e a r e n ' t a n y d i r e c t o r r e l a t e d c h a n g e s t h e n e a r l y - o u t
guard ! directChanges . isEmpty || ! relatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
return ( dataCache , pageInfo , false , getAssociatedDataInfo ( db , pageInfo ) )
}
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : ( updatedPageInfo . totalCount + dataSizeDiff )
)
}
}
// I f t h e r e a r e n o i n s e r t e d / u p d a t e d r o w s t h e n t r i g g e r t h e u p d a t e c a l l b a c k a n d s t o p h e r e
let changesToQuery : [ PagedData . TrackedChange ] = directChanges
. filter { $0 . kind != . delete }
guard ! changesToQuery . isEmpty || ! relatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
updateDataAndCallbackIfNeeded ( updatedDataCache , updatedPageInfo , ! deletionChanges . isEmpty )
return
}
// F i r s t w e n e e d t o g e t t h e r o w I d s f o r t h e p a g e d d a t a c o n n e c t e d t o a n y o f t h e r e l a t e d c h a n g e s
let pagedRowIdsForRelatedChanges : Set < Int64 > = {
guard ! relatedChanges . isEmpty else { return [ ] }
return relatedChanges
. reduce ( into : [ ] ) { result , next in
guard
let observedChange : PagedData . ObservedChanges = observedTableChangeTypes [ next . key ] ,
let joinToPagedType : SQL = observedChange . joinToPagedType
else { return }
// S t o r e a m u t a b l e c o p i e s o f t h e d a t a C a c h e a n d p a g e I n f o f o r u p d a t i n g
var updatedDataCache : DataCache < T > = dataCache
var updatedPageInfo : PagedData . PageInfo = pageInfo
let deletionChanges : [ Int64 ] = directChanges
. filter { $0 . kind = = . delete }
. map { $0 . rowId }
let oldDataCount : Int = dataCache . count
// F i r s t r e m o v e a n y i t e m s w h i c h h a v e b e e n d e l e t e d
if ! deletionChanges . isEmpty {
updatedDataCache = updatedDataCache . deleting ( rowIds : deletionChanges )
let pagedRowIds : [ Int64 ] = PagedData . pagedRowIdsForRelatedRowIds (
db ,
tableName : next . key ,
pagedTableName : pagedTableName ,
relatedRowIds : Array ( next . value . map { $0 . rowId } . asSet ( ) ) ,
joinToPagedType : joinToPagedType
)
// M a k e s u r e t h e r e w e r e a c t u a l l y c h a n g e s
if updatedDataCache . count != oldDataCount {
let dataSizeDiff : Int = ( updatedDataCache . count - oldDataCount )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : ( updatedPageInfo . totalCount + dataSizeDiff )
)
}
}
// I f t h e r e a r e n o i n s e r t e d / u p d a t e d r o w s t h e n t r i g g e r t h e n e a r l y - o u t
let changesToQuery : [ PagedData . TrackedChange ] = directChanges
. filter { $0 . kind != . delete }
guard ! changesToQuery . isEmpty || ! relatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
let associatedData : AssociatedDataInfo = getAssociatedDataInfo ( db , updatedPageInfo )
return ( updatedDataCache , updatedPageInfo , ! deletionChanges . isEmpty , associatedData )
}
// N e x t w e n e e d t o d e t e r m i n e i f a n y r e l a t e d c h a n g e s w e r e a s s o c i a t e d t o t h e p a g e d D a t a w e a r e
// o b s e r v i n g , i f t h e y a r e n ' t ( a n d t h e r e w e r e n o o t h e r d i r e c t c h a n g e s ) w e c a n e a r l y - o u t
let pagedRowIdsForRelatedChanges : Set < Int64 > = {
guard ! relatedChanges . isEmpty else { return [ ] }
result . append ( contentsOf : pagedRowIds )
return relatedChanges
. reduce ( into : [ ] ) { result , next in
guard
let observedChange : PagedData . ObservedChanges = observedTableChangeTypes [ next . key ] ,
let joinToPagedType : SQL = observedChange . joinToPagedType
else { return }
let pagedRowIds : [ Int64 ] = PagedData . pagedRowIdsForRelatedRowIds (
db ,
tableName : next . key ,
pagedTableName : pagedTableName ,
relatedRowIds : Array ( next . value . map { $0 . rowId } . asSet ( ) ) ,
joinToPagedType : joinToPagedType
)
result . append ( contentsOf : pagedRowIds )
}
. asSet ( )
} ( )
guard ! changesToQuery . isEmpty || ! pagedRowIdsForRelatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
let associatedData : AssociatedDataInfo = getAssociatedDataInfo ( db , updatedPageInfo )
return ( updatedDataCache , updatedPageInfo , ! deletionChanges . isEmpty , associatedData )
}
. asSet ( )
} ( )
guard ! changesToQuery . isEmpty || ! pagedRowIdsForRelatedChanges . isEmpty || ! relatedDeletions . isEmpty else {
updateDataAndCallbackIfNeeded ( updatedDataCache , updatedPageInfo , ! deletionChanges . isEmpty )
return
}
// F e t c h t h e i n d e x e s o f t h e r o w I d s s o w e c a n d e t e r m i n e w h e t h e r t h e y s h o u l d b e a d d e d t o t h e s c r e e n
let directRowIds : Set < Int64 > = changesToQuery . map { $0 . rowId } . asSet ( )
let pagedRowIdsForRelatedDeletions : Set < Int64 > = relatedDeletions
. compactMap { $0 . pagedRowIdsForRelatedDeletion }
. flatMap { $0 }
. asSet ( )
let itemIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( directRowIds ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
let relatedChangeIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( pagedRowIdsForRelatedChanges ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
let relatedDeletionIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( pagedRowIdsForRelatedDeletions ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
// D e t e r m i n e i f t h e i n d e x e s f o r t h e r o w i d s s h o u l d b e d i s p l a y e d o n t h e s c r e e n a n d r e m o v e a n y
// w h i c h s h o u l d n ' t - v a l u e s l e s s t h a n ' c u r r e n t C o u n t ' o r i f t h e r e i s a t l e a s t o n e v a l u e l e s s t h a n
// ' c u r r e n t C o u n t ' a n d t h e i n d e x e s a r e s e q u e n t i a l ( i e . m o r e t h a n t h e c u r r e n t l o a d e d c o n t e n t w a s
// a d d e d a t o n c e )
func determineValidChanges ( for indexInfo : [ PagedData . RowIndexInfo ] ) -> [ Int64 ] {
let indexes : [ Int64 ] = Array ( indexInfo
. map { $0 . rowIndex }
. sorted ( )
. asSet ( ) )
let indexesAreSequential : Bool = ( indexes . map { $0 - 1 } . dropFirst ( ) = = indexes . dropLast ( ) )
let hasOneValidIndex : Bool = indexInfo . contains ( where : { info -> Bool in
info . rowIndex >= updatedPageInfo . pageOffset && (
info . rowIndex < updatedPageInfo . currentCount || (
updatedPageInfo . currentCount < updatedPageInfo . pageSize &&
info . rowIndex <= ( updatedPageInfo . pageOffset + updatedPageInfo . pageSize )
)
// F e t c h t h e i n d e x e s o f t h e r o w I d s s o w e c a n d e t e r m i n e w h e t h e r t h e y s h o u l d b e a d d e d t o t h e s c r e e n
let directRowIds : Set < Int64 > = changesToQuery . map { $0 . rowId } . asSet ( )
let pagedRowIdsForRelatedDeletions : Set < Int64 > = relatedDeletions
. compactMap { $0 . pagedRowIdsForRelatedDeletion }
. flatMap { $0 }
. asSet ( )
let itemIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( directRowIds ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
} )
return ( indexesAreSequential && hasOneValidIndex ?
indexInfo . map { $0 . rowId } :
indexInfo
. filter { info -> Bool in
let relatedChangeIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( pagedRowIdsForRelatedChanges ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
let relatedDeletionIndexes : [ PagedData . RowIndexInfo ] = PagedData . indexes (
db ,
rowIds : Array ( pagedRowIdsForRelatedDeletions ) ,
tableName : pagedTableName ,
requiredJoinSQL : joinSQL ,
orderSQL : orderSQL ,
filterSQL : filterSQL
)
// D e t e r m i n e i f t h e i n d e x e s f o r t h e r o w i d s s h o u l d b e d i s p l a y e d o n t h e s c r e e n a n d r e m o v e a n y
// w h i c h s h o u l d n ' t - v a l u e s l e s s t h a n ' c u r r e n t C o u n t ' o r i f t h e r e i s a t l e a s t o n e v a l u e l e s s t h a n
// ' c u r r e n t C o u n t ' a n d t h e i n d e x e s a r e s e q u e n t i a l ( i e . m o r e t h a n t h e c u r r e n t l o a d e d c o n t e n t w a s
// a d d e d a t o n c e )
func determineValidChanges ( for indexInfo : [ PagedData . RowIndexInfo ] ) -> [ Int64 ] {
let indexes : [ Int64 ] = Array ( indexInfo
. map { $0 . rowIndex }
. sorted ( )
. asSet ( ) )
let indexesAreSequential : Bool = ( indexes . map { $0 - 1 } . dropFirst ( ) = = indexes . dropLast ( ) )
let hasOneValidIndex : Bool = indexInfo . contains ( where : { info -> Bool in
info . rowIndex >= updatedPageInfo . pageOffset && (
info . rowIndex < updatedPageInfo . currentCount || (
updatedPageInfo . currentCount < updatedPageInfo . pageSize &&
info . rowIndex <= ( updatedPageInfo . pageOffset + updatedPageInfo . pageSize )
)
)
} )
return ( indexesAreSequential && hasOneValidIndex ?
indexInfo . map { $0 . rowId } :
indexInfo
. filter { info -> Bool in
info . rowIndex >= updatedPageInfo . pageOffset && (
info . rowIndex < updatedPageInfo . currentCount || (
updatedPageInfo . currentCount < updatedPageInfo . pageSize &&
info . rowIndex <= ( updatedPageInfo . pageOffset + updatedPageInfo . pageSize )
)
)
}
. map { info -> Int64 in info . rowId }
)
}
let validChangeRowIds : [ Int64 ] = determineValidChanges ( for : itemIndexes )
let validRelatedChangeRowIds : [ Int64 ] = determineValidChanges ( for : relatedChangeIndexes )
let validRelatedDeletionRowIds : [ Int64 ] = determineValidChanges ( for : relatedDeletionIndexes )
let countBefore : Int = itemIndexes . filter { $0 . rowIndex < updatedPageInfo . pageOffset } . count
// I f t h e n u m b e r o f i n d e x e s d o e s n ' t m a t c h t h e n u m b e r o f r o w I d s t h e n i t m e a n s s o m e t h i n g c h a n g e d
// r e s u l t i n g i n a n i t e m b e i n g f i l t e r e d o u t
func performRemovalsIfNeeded ( for rowIds : Set < Int64 > , indexes : [ PagedData . RowIndexInfo ] ) {
let uniqueIndexes : Set < Int64 > = indexes . map { $0 . rowId } . asSet ( )
// I f t h e y h a v e t h e s a m e c o u n t t h e n n o t h i n w a s f i l t e r e d o u t s o d o n o t h i n g
guard rowIds . count != uniqueIndexes . count else { return }
// O t h e r w i s e s o m e t h i n g w a s p r o b a b l y r e m o v e d s o t r y t o r e m o v e i t f r o m t h e c a c h e
let rowIdsRemoved : Set < Int64 > = rowIds . subtracting ( uniqueIndexes )
let preDeletionCount : Int = updatedDataCache . count
updatedDataCache = updatedDataCache . deleting ( rowIds : Array ( rowIdsRemoved ) )
// L a s t l y m a k e s u r e t h e r e w e r e a c t u a l l y c h a n g e s b e f o r e u p d a t i n g t h e p a g e i n f o
guard updatedDataCache . count != preDeletionCount else { return }
let dataSizeDiff : Int = ( updatedDataCache . count - preDeletionCount )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : ( updatedPageInfo . totalCount + dataSizeDiff )
)
}
// A c t u a l l y p e r f o r m a n y r e q u i r e d r e m o v a l s
performRemovalsIfNeeded ( for : directRowIds , indexes : itemIndexes )
performRemovalsIfNeeded ( for : pagedRowIdsForRelatedChanges , indexes : relatedChangeIndexes )
performRemovalsIfNeeded ( for : pagedRowIdsForRelatedDeletions , indexes : relatedDeletionIndexes )
// U p d a t e t h e o f f s e t a n d t o t a l C o u n t e v e n i f t h e r o w s a r e o u t s i d e o f t h e c u r r e n t p a g e ( n e e d t o
// i n o r d e r t o e n s u r e t h e ' l o a d m o r e ' s e c t i o n s a r e a c c u r a t e )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : ( updatedPageInfo . pageOffset + countBefore ) ,
currentCount : updatedPageInfo . currentCount ,
totalCount : (
updatedPageInfo . totalCount +
changesToQuery
. filter { $0 . kind = = . insert }
. filter { validChangeRowIds . contains ( $0 . rowId ) }
. count
)
)
// I f t h e r e a r e n o v a l i d r o w i d s t h e n e a r l y - o u t ( a t t h i s p o i n t t h e p a g e I n f o w o u l d h a v e c h a n g e d
// s o w e w a n t t o f l a t ' h a s C h a n g e s ' a s t r u e )
guard ! validChangeRowIds . isEmpty || ! validRelatedChangeRowIds . isEmpty || ! validRelatedDeletionRowIds . isEmpty else {
let associatedData : AssociatedDataInfo = getAssociatedDataInfo ( db , updatedPageInfo )
return ( updatedDataCache , updatedPageInfo , true , associatedData )
}
// F e t c h t h e i n s e r t e d / u p d a t e d r o w s
let targetRowIds : [ Int64 ] = Array ( ( validChangeRowIds + validRelatedChangeRowIds + validRelatedDeletionRowIds ) . asSet ( ) )
let updatedItems : [ T ] = {
do { return try dataQuery ( targetRowIds ) . fetchAll ( db ) }
catch {
SNLog ( " [PagedDatabaseObserver] Error fetching data during change: \( error ) " )
return [ ]
}
. map { info -> Int64 in info . rowId }
)
} ( )
updatedDataCache = updatedDataCache . upserting ( items : updatedItems )
// U p d a t e t h e c u r r e n t C o u n t f o r t h e u p s e r t e d d a t a
let dataSizeDiff : Int = ( updatedDataCache . count - oldDataCount )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : updatedPageInfo . totalCount
)
// R e t u r n t h e f i n a l u p d a t e d d a t a
let associatedData : AssociatedDataInfo = getAssociatedDataInfo ( db , updatedPageInfo )
return ( updatedDataCache , updatedPageInfo , true , associatedData )
}
. defaulting ( to : ( cache : dataCache , pageInfo : pageInfo , hasChanges : false , associatedData : [ ] ) )
// N o w t h a t w e h a v e a l l o f t h e c h a n g e s , c h e c k i f t h e r e w e r e a c t u a l l y a n y c h a n g e s
guard updatedData . hasChanges || updatedData . associatedData . contains ( where : { hasChanges , _ in hasChanges } ) else {
return
}
let validChangeRowIds : [ Int64 ] = determineValidChanges ( for : itemIndexes )
let validRelatedChangeRowIds : [ Int64 ] = determineValidChanges ( for : relatedChangeIndexes )
let validRelatedDeletionRowIds : [ Int64 ] = determineValidChanges ( for : relatedDeletionIndexes )
let countBefore : Int = itemIndexes . filter { $0 . rowIndex < updatedPageInfo . pageOffset } . count
// I f t h e n u m b e r o f i n d e x e s d o e s n ' t m a t c h t h e n u m b e r o f r o w I d s t h e n i t m e a n s s o m e t h i n g c h a n g e d
// r e s u l t i n g i n a n i t e m b e i n g f i l t e r e d o u t
func performRemovalsIfNeeded ( for rowIds : Set < Int64 > , indexes : [ PagedData . RowIndexInfo ] ) {
let uniqueIndexes : Set < Int64 > = indexes . map { $0 . rowId } . asSet ( )
// I f t h e y h a v e t h e s a m e c o u n t t h e n n o t h i n w a s f i l t e r e d o u t s o d o n o t h i n g
guard rowIds . count != uniqueIndexes . count else { return }
// O t h e r w i s e s o m e t h i n g w a s p r o b a b l y r e m o v e d s o t r y t o r e m o v e i t f r o m t h e c a c h e
let rowIdsRemoved : Set < Int64 > = rowIds . subtracting ( uniqueIndexes )
let preDeletionCount : Int = updatedDataCache . count
updatedDataCache = updatedDataCache . deleting ( rowIds : Array ( rowIdsRemoved ) )
// I f t h e a s s o c i a t e d d a t a c h a n g e d t h e n u p d a t e t h e u p d a t e d C a c h e d D a t a w i t h t h e u p d a t e d a s s o c i a t e d d a t a
var finalUpdatedDataCache : DataCache < T > = updatedData . cache
// L a s t l y m a k e s u r e t h e r e w e r e a c t u a l l y c h a n g e s b e f o r e u p d a t i n g t h e p a g e i n f o
guard updatedDataCache . count != preDeletionCount else { return }
let dataSizeDiff : Int = ( updatedDataCache . count - preDeletionCount )
updatedData . associatedData . forEach { hasChanges , associatedData in
guard updatedData . hasChanges || hasChanges else { return }
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : ( updatedPageInfo . totalCount + dataSizeDiff )
)
finalUpdatedDataCache = associatedData . updateAssociatedData ( to : finalUpdatedDataCache )
}
// A c t u a l l y p e r f o r m a n y r e q u i r e d r e m o v a l s
performRemovalsIfNeeded ( for : directRowIds , indexes : itemIndexes )
performRemovalsIfNeeded ( for : pagedRowIdsForRelatedChanges , indexes : relatedChangeIndexes )
performRemovalsIfNeeded ( for : pagedRowIdsForRelatedDeletions , indexes : relatedDeletionIndexes )
// U p d a t e t h e o f f s e t a n d t o t a l C o u n t e v e n i f t h e r o w s a r e o u t s i d e o f t h e c u r r e n t p a g e ( n e e d t o
// i n o r d e r t o e n s u r e t h e ' l o a d m o r e ' s e c t i o n s a r e a c c u r a t e )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : ( updatedPageInfo . pageOffset + countBefore ) ,
currentCount : updatedPageInfo . currentCount ,
totalCount : (
updatedPageInfo . totalCount +
changesToQuery
. filter { $0 . kind = = . insert }
. filter { validChangeRowIds . contains ( $0 . rowId ) }
. count
)
)
// I f t h e r e a r e n o v a l i d r o w i d s t h e n s t o p h e r e ( t r i g g e r u p d a t e s t h o u g h s i n c e t h e p a g e i n f o
// h a s c h a n g e s )
guard ! validChangeRowIds . isEmpty || ! validRelatedChangeRowIds . isEmpty || ! validRelatedDeletionRowIds . isEmpty else {
updateDataAndCallbackIfNeeded ( updatedDataCache , updatedPageInfo , true )
return
}
// F e t c h t h e i n s e r t e d / u p d a t e d r o w s
let targetRowIds : [ Int64 ] = Array ( ( validChangeRowIds + validRelatedChangeRowIds + validRelatedDeletionRowIds ) . asSet ( ) )
let updatedItems : [ T ] = ( try ? dataQuery ( targetRowIds )
. fetchAll ( db ) )
. defaulting ( to : [ ] )
// U p d a t e t h e c a c h e , p a g e I n f o a n d t h e c h a n g e c a l l b a c k
self . dataCache . mutate { $0 = finalUpdatedDataCache }
self . pageInfo . mutate { $0 = updatedData . pageInfo }
// P r o c e s s t h e u p s e r t e d d a t a
updatedDataCache = updatedDataCache . upserting ( items : updatedItems )
// U p d a t e t h e c u r r e n t C o u n t f o r t h e u p s e r t e d d a t a
let dataSizeDiff : Int = ( updatedDataCache . count - oldDataCount )
updatedPageInfo = PagedData . PageInfo (
pageSize : updatedPageInfo . pageSize ,
pageOffset : updatedPageInfo . pageOffset ,
currentCount : ( updatedPageInfo . currentCount + dataSizeDiff ) ,
totalCount : updatedPageInfo . totalCount
)
updateDataAndCallbackIfNeeded ( updatedDataCache , updatedPageInfo , true )
// T r i g g e r t h e u n s o r t e d c h a n g e c a l l b a c k ( t h e a c t u a l U I u p d a t e t r i g g e r i n g s h o u l d e v e n t u a l l y b e r u n o n
// t h e m a i n t h r e a d v i a t h e ` P a g e d D a t a . p r o c e s s A n d T r i g g e r U p d a t e s ` f u n c t i o n )
self . onChangeUnsorted ( finalUpdatedDataCache . values , updatedData . pageInfo )
}
public func databaseDidRollback ( _ db : Database ) { }
Morgan Pretty
2 years ago