diff options
Diffstat (limited to 'dsa/FireBase/Offline')
| -rw-r--r-- | dsa/FireBase/Offline/ConcurrentOfflineDatabase.cs | 233 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/DatabaseExtensions.cs | 257 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/ISetHandler.cs | 10 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/InitialPullStrategy.cs | 23 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/Internals/MemberAccessVisitor.cs | 46 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/OfflineCacheAdapter.cs | 152 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/OfflineDatabase.cs | 228 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/OfflineEntry.cs | 99 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/RealtimeDatabase.cs | 479 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/SetHandler.cs | 19 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/StreamingOptions.cs | 23 | ||||
| -rw-r--r-- | dsa/FireBase/Offline/SyncOptions.cs | 28 |
12 files changed, 1597 insertions, 0 deletions
diff --git a/dsa/FireBase/Offline/ConcurrentOfflineDatabase.cs b/dsa/FireBase/Offline/ConcurrentOfflineDatabase.cs new file mode 100644 index 0000000..1a9e607 --- /dev/null +++ b/dsa/FireBase/Offline/ConcurrentOfflineDatabase.cs @@ -0,0 +1,233 @@ +using System; +using System.Collections; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using LiteDB; + +namespace Firebase.Database.Offline +{ + /// <summary> + /// The offline database. + /// </summary> + public class ConcurrentOfflineDatabase : IDictionary<string, OfflineEntry> + { + private readonly ConcurrentDictionary<string, OfflineEntry> ccache; + private readonly LiteRepository db; + + /// <summary> + /// Initializes a new instance of the <see cref="OfflineDatabase" /> class. + /// </summary> + /// <param name="itemType"> The item type which is used to determine the database file name. </param> + /// <param name="filenameModifier"> Custom string which will get appended to the file name. </param> + public ConcurrentOfflineDatabase(Type itemType, string filenameModifier) + { + var fullName = GetFileName(itemType.ToString()); + if (fullName.Length > 100) fullName = fullName.Substring(0, 100); + + var mapper = BsonMapper.Global; + mapper.Entity<OfflineEntry>().Id(o => o.Key); + + var root = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData); + var filename = fullName + filenameModifier + ".db"; + var path = Path.Combine(root, filename); + db = new LiteRepository(new LiteDatabase(path, mapper)); + + var cache = db.Database + .GetCollection<OfflineEntry>() + .FindAll() + .ToDictionary(o => o.Key, o => o); + + ccache = new ConcurrentDictionary<string, OfflineEntry>(cache); + } + + /// <summary> + /// Gets the number of elements contained in the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <returns> The number of elements contained in the <see cref="T:System.Collections.Generic.ICollection`1" />. </returns> + public int Count => ccache.Count; + + /// <summary> + /// Gets a value indicating whether this is a read-only collection. + /// </summary> + public bool IsReadOnly => false; + + /// <summary> + /// Gets an <see cref="T:System.Collections.Generic.ICollection`1" /> containing the keys of the + /// <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <returns> + /// An <see cref="T:System.Collections.Generic.ICollection`1" /> containing the keys of the object that + /// implements <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public ICollection<string> Keys => ccache.Keys; + + /// <summary> + /// Gets an <see cref="T:System.Collections.Generic.ICollection`1" /> containing the values in the + /// <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <returns> + /// An <see cref="T:System.Collections.Generic.ICollection`1" /> containing the values in the object that + /// implements <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public ICollection<OfflineEntry> Values => ccache.Values; + + /// <summary> + /// Gets or sets the element with the specified key. + /// </summary> + /// <param name="key">The key of the element to get or set.</param> + /// <returns> The element with the specified key. </returns> + public OfflineEntry this[string key] + { + get => ccache[key]; + + set + { + ccache.AddOrUpdate(key, value, (k, existing) => value); + db.Upsert(value); + } + } + + /// <summary> + /// Returns an enumerator that iterates through the collection. + /// </summary> + /// <returns> An enumerator that can be used to iterate through the collection. </returns> + public IEnumerator<KeyValuePair<string, OfflineEntry>> GetEnumerator() + { + return ccache.GetEnumerator(); + } + + IEnumerator IEnumerable.GetEnumerator() + { + return GetEnumerator(); + } + + /// <summary> + /// Adds an item to the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <param name="item">The object to add to the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + public void Add(KeyValuePair<string, OfflineEntry> item) + { + Add(item.Key, item.Value); + } + + /// <summary> + /// Removes all items from the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + public void Clear() + { + ccache.Clear(); + db.Delete<OfflineEntry>(LiteDB.Query.All()); + } + + /// <summary> + /// Determines whether the <see cref="T:System.Collections.Generic.ICollection`1" /> contains a specific value. + /// </summary> + /// <param name="item">The object to locate in the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + /// <returns> + /// True if <paramref name="item" /> is found in the <see cref="T:System.Collections.Generic.ICollection`1" />; + /// otherwise, false. + /// </returns> + public bool Contains(KeyValuePair<string, OfflineEntry> item) + { + return ContainsKey(item.Key); + } + + /// <summary> + /// Copies the elements of the <see cref="T:System.Collections.Generic.ICollection`1" /> to an + /// <see cref="T:System.Array" />, starting at a particular <see cref="T:System.Array" /> index. + /// </summary> + /// <param name="array"> + /// The one-dimensional <see cref="T:System.Array" /> that is the destination of the elements copied + /// from <see cref="T:System.Collections.Generic.ICollection`1" />. The <see cref="T:System.Array" /> must have + /// zero-based indexing. + /// </param> + /// <param name="arrayIndex">The zero-based index in <paramref name="array" /> at which copying begins.</param> + public void CopyTo(KeyValuePair<string, OfflineEntry>[] array, int arrayIndex) + { + ccache.ToList().CopyTo(array, arrayIndex); + } + + /// <summary> + /// Removes the first occurrence of a specific object from the + /// <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <param name="item">The object to remove from the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + /// <returns> + /// True if <paramref name="item" /> was successfully removed from the + /// <see cref="T:System.Collections.Generic.ICollection`1" />; otherwise, false. This method also returns false if + /// <paramref name="item" /> is not found in the original <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </returns> + public bool Remove(KeyValuePair<string, OfflineEntry> item) + { + return Remove(item.Key); + } + + /// <summary> + /// Determines whether the <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an element with the + /// specified key. + /// </summary> + /// <param name="key">The key to locate in the <see cref="T:System.Collections.Generic.IDictionary`2" />.</param> + /// <returns> + /// True if the <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an element with the key; + /// otherwise, false. + /// </returns> + public bool ContainsKey(string key) + { + return ccache.ContainsKey(key); + } + + /// <summary> + /// Adds an element with the provided key and value to the <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <param name="key">The object to use as the key of the element to add.</param> + /// <param name="value">The object to use as the value of the element to add.</param> + public void Add(string key, OfflineEntry value) + { + ccache.AddOrUpdate(key, value, (k, existing) => value); + db.Upsert(value); + } + + /// <summary> + /// Removes the element with the specified key from the <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <param name="key">The key of the element to remove.</param> + /// <returns> + /// True if the element is successfully removed; otherwise, false. This method also returns false if + /// <paramref name="key" /> was not found in the original <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public bool Remove(string key) + { + ccache.TryRemove(key, out _); + return db.Delete<OfflineEntry>(key); + } + + /// <summary> + /// Gets the value associated with the specified key. + /// </summary> + /// <param name="key">The key whose value to get.</param> + /// <param name="value"> + /// When this method returns, the value associated with the specified key, if the key is found; + /// otherwise, the default value for the type of the <paramref name="value" /> parameter. This parameter is passed + /// uninitialized. + /// </param> + /// <returns> + /// True if the object that implements <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an + /// element with the specified key; otherwise, false. + /// </returns> + public bool TryGetValue(string key, out OfflineEntry value) + { + return ccache.TryGetValue(key, out value); + } + + private string GetFileName(string fileName) + { + var invalidChars = new[] {'`', '[', ',', '='}; + foreach (var c in invalidChars.Concat(Path.GetInvalidFileNameChars()).Distinct()) + fileName = fileName.Replace(c, '_'); + + return fileName; + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/DatabaseExtensions.cs b/dsa/FireBase/Offline/DatabaseExtensions.cs new file mode 100644 index 0000000..e7c4074 --- /dev/null +++ b/dsa/FireBase/Offline/DatabaseExtensions.cs @@ -0,0 +1,257 @@ +using System; +using System.Collections.Generic; +using System.Linq.Expressions; +using System.Reflection; +using Firebase.Database.Query; + +namespace Firebase.Database.Offline +{ + public static class DatabaseExtensions + { + /// <summary> + /// Create new instances of the <see cref="RealtimeDatabase{T}" />. + /// </summary> + /// <typeparam name="T"> Type of elements. </typeparam> + /// <param name="filenameModifier"> Custom string which will get appended to the file name. </param> + /// <param name="elementRoot"> Optional custom root element of received json items. </param> + /// <param name="streamingOptions"> Realtime streaming options. </param> + /// <param name="initialPullStrategy"> Specifies what strategy should be used for initial pulling of server data. </param> + /// <param name="pushChanges"> + /// Specifies whether changed items should actually be pushed to the server. It this is false, + /// then Put / Post / Delete will not affect server data. + /// </param> + /// <returns> The <see cref="RealtimeDatabase{T}" />. </returns> + public static RealtimeDatabase<T> AsRealtimeDatabase<T>(this ChildQuery query, string filenameModifier = "", + string elementRoot = "", StreamingOptions streamingOptions = StreamingOptions.LatestOnly, + InitialPullStrategy initialPullStrategy = InitialPullStrategy.MissingOnly, bool pushChanges = true) + where T : class + { + return new RealtimeDatabase<T>(query, elementRoot, query.Client.Options.OfflineDatabaseFactory, + filenameModifier, streamingOptions, initialPullStrategy, pushChanges); + } + + /// <summary> + /// Create new instances of the <see cref="RealtimeDatabase{T}" />. + /// </summary> + /// <typeparam name="T"> Type of elements. </typeparam> + /// <typeparam name="TSetHandler"> Type of the custom <see cref="ISetHandler{T}" /> to use. </typeparam> + /// <param name="filenameModifier"> Custom string which will get appended to the file name. </param> + /// <param name="elementRoot"> Optional custom root element of received json items. </param> + /// <param name="streamingOptions"> Realtime streaming options. </param> + /// <param name="initialPullStrategy"> Specifies what strategy should be used for initial pulling of server data. </param> + /// <param name="pushChanges"> + /// Specifies whether changed items should actually be pushed to the server. It this is false, + /// then Put / Post / Delete will not affect server data. + /// </param> + /// <returns> The <see cref="RealtimeDatabase{T}" />. </returns> + public static RealtimeDatabase<T> AsRealtimeDatabase<T, TSetHandler>(this ChildQuery query, + string filenameModifier = "", string elementRoot = "", + StreamingOptions streamingOptions = StreamingOptions.LatestOnly, + InitialPullStrategy initialPullStrategy = InitialPullStrategy.MissingOnly, bool pushChanges = true) + where T : class + where TSetHandler : ISetHandler<T>, new() + { + return new RealtimeDatabase<T>(query, elementRoot, query.Client.Options.OfflineDatabaseFactory, + filenameModifier, streamingOptions, initialPullStrategy, pushChanges, + Activator.CreateInstance<TSetHandler>()); + } + + /// <summary> + /// Overwrites existing object with given key leaving any missing properties intact in firebase. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="obj"> The object to set. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Patch<T>(this RealtimeDatabase<T> db, string key, T obj, bool syncOnline = true, + int priority = 1) + where T : class + { + db.Set(key, obj, syncOnline ? SyncOptions.Patch : SyncOptions.None, priority); + } + + /// <summary> + /// Overwrites existing object with given key. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="obj"> The object to set. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Put<T>(this RealtimeDatabase<T> db, string key, T obj, bool syncOnline = true, + int priority = 1) + where T : class + { + db.Set(key, obj, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + + /// <summary> + /// Adds a new entity to the Database. + /// </summary> + /// <param name="obj"> The object to add. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + /// <returns> The generated key for this object. </returns> + public static string Post<T>(this RealtimeDatabase<T> db, T obj, bool syncOnline = true, int priority = 1) + where T : class + { + var key = FirebaseKeyGenerator.Next(); + + db.Set(key, obj, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + + return key; + } + + /// <summary> + /// Deletes the entity with the given key. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Delete<T>(this RealtimeDatabase<T> db, string key, bool syncOnline = true, int priority = 1) + where T : class + { + db.Set(key, null, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + + /// <summary> + /// Do a Put for a nested property specified by <paramref name="propertyExpression" /> of an object with key + /// <paramref name="key" />. + /// </summary> + /// <typeparam name="T"> Type of the root elements. </typeparam> + /// <typeparam name="TProperty"> Type of the property being modified</typeparam> + /// <param name="db"> Database instance. </param> + /// <param name="key"> Key of the root element to modify. </param> + /// <param name="propertyExpression"> Expression on the root element leading to target value to modify. </param> + /// <param name="value"> Value to put. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Put<T, TProperty>(this RealtimeDatabase<T> db, string key, + Expression<Func<T, TProperty>> propertyExpression, TProperty value, bool syncOnline = true, + int priority = 1) + where T : class + { + db.Set(key, propertyExpression, value, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + + /// <summary> + /// Do a Patch for a nested property specified by <paramref name="propertyExpression" /> of an object with key + /// <paramref name="key" />. + /// </summary> + /// <typeparam name="T"> Type of the root elements. </typeparam> + /// <typeparam name="TProperty"> Type of the property being modified</typeparam> + /// <param name="db"> Database instance. </param> + /// <param name="key"> Key of the root element to modify. </param> + /// <param name="propertyExpression"> Expression on the root element leading to target value to modify. </param> + /// <param name="value"> Value to patch. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Patch<T, TProperty>(this RealtimeDatabase<T> db, string key, + Expression<Func<T, TProperty>> propertyExpression, TProperty value, bool syncOnline = true, + int priority = 1) + where T : class + { + db.Set(key, propertyExpression, value, syncOnline ? SyncOptions.Patch : SyncOptions.None, priority); + } + + /// <summary> + /// Delete a nested property specified by <paramref name="propertyExpression" /> of an object with key + /// <paramref name="key" />. This basically does a Put with null value. + /// </summary> + /// <typeparam name="T"> Type of the root elements. </typeparam> + /// <typeparam name="TProperty"> Type of the property being modified</typeparam> + /// <param name="db"> Database instance. </param> + /// <param name="key"> Key of the root element to modify. </param> + /// <param name="propertyExpression"> Expression on the root element leading to target value to modify. </param> + /// <param name="value"> Value to put. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Delete<T, TProperty>(this RealtimeDatabase<T> db, string key, + Expression<Func<T, TProperty>> propertyExpression, bool syncOnline = true, int priority = 1) + where T : class + where TProperty : class + { + db.Set(key, propertyExpression, null, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + + /// <summary> + /// Post a new entity into the nested dictionary specified by <paramref name="propertyExpression" /> of an object with + /// key <paramref name="key" />. + /// The key of the new entity is automatically generated. + /// </summary> + /// <typeparam name="T"> Type of the root elements. </typeparam> + /// <typeparam name="TSelector"> Type of the dictionary being modified</typeparam> + /// <typeparam name="TProperty"> Type of the value within the dictionary being modified</typeparam> + /// <param name="db"> Database instance. </param> + /// <param name="key"> Key of the root element to modify. </param> + /// <param name="propertyExpression"> Expression on the root element leading to target value to modify. </param> + /// <param name="value"> Value to put. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Post<T, TSelector, TProperty>(this RealtimeDatabase<T> db, string key, + Expression<Func<T, TSelector>> propertyExpression, TProperty value, bool syncOnline = true, + int priority = 1) + where T : class + where TSelector : IDictionary<string, TProperty> + { + var nextKey = FirebaseKeyGenerator.Next(); + var expression = Expression.Lambda<Func<T, TProperty>>( + Expression.Call(propertyExpression.Body, + typeof(TSelector).GetRuntimeMethod("get_Item", new[] {typeof(string)}), + Expression.Constant(nextKey)), propertyExpression.Parameters); + db.Set(key, expression, value, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + + /// <summary> + /// Delete an entity with key <paramref name="dictionaryKey" /> in the nested dictionary specified by + /// <paramref name="propertyExpression" /> of an object with key <paramref name="key" />. + /// The key of the new entity is automatically generated. + /// </summary> + /// <typeparam name="T"> Type of the root elements. </typeparam> + /// <typeparam name="TSelector"> Type of the dictionary being modified</typeparam> + /// <typeparam name="TProperty"> Type of the value within the dictionary being modified</typeparam> + /// <param name="db"> Database instance. </param> + /// <param name="key"> Key of the root element to modify. </param> + /// <param name="propertyExpression"> Expression on the root element leading to target value to modify. </param> + /// <param name="dictionaryKey"> Key within the nested dictionary to delete. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public static void Delete<T, TProperty>(this RealtimeDatabase<T> db, string key, + Expression<Func<T, IDictionary<string, TProperty>>> propertyExpression, string dictionaryKey, + bool syncOnline = true, int priority = 1) + where T : class + { + var expression = Expression.Lambda<Func<T, TProperty>>( + Expression.Call(propertyExpression.Body, + typeof(IDictionary<string, TProperty>).GetRuntimeMethod("get_Item", new[] {typeof(string)}), + Expression.Constant(dictionaryKey)), propertyExpression.Parameters); + db.Set(key, expression, null, syncOnline ? SyncOptions.Put : SyncOptions.None, priority); + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/ISetHandler.cs b/dsa/FireBase/Offline/ISetHandler.cs new file mode 100644 index 0000000..c04bd41 --- /dev/null +++ b/dsa/FireBase/Offline/ISetHandler.cs @@ -0,0 +1,10 @@ +using System.Threading.Tasks; +using Firebase.Database.Query; + +namespace Firebase.Database.Offline +{ + public interface ISetHandler<in T> + { + Task SetAsync(ChildQuery query, string key, OfflineEntry entry); + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/InitialPullStrategy.cs b/dsa/FireBase/Offline/InitialPullStrategy.cs new file mode 100644 index 0000000..ca2bebf --- /dev/null +++ b/dsa/FireBase/Offline/InitialPullStrategy.cs @@ -0,0 +1,23 @@ +namespace Firebase.Database.Offline +{ + /// <summary> + /// Specifies the strategy for initial pull of server data. + /// </summary> + public enum InitialPullStrategy + { + /// <summary> + /// Don't pull anything. + /// </summary> + None, + + /// <summary> + /// Pull only what isn't already stored offline. + /// </summary> + MissingOnly, + + /// <summary> + /// Pull everything that exists on the server. + /// </summary> + Everything + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/Internals/MemberAccessVisitor.cs b/dsa/FireBase/Offline/Internals/MemberAccessVisitor.cs new file mode 100644 index 0000000..89a77da --- /dev/null +++ b/dsa/FireBase/Offline/Internals/MemberAccessVisitor.cs @@ -0,0 +1,46 @@ +using System.Collections.Generic; +using System.Linq.Expressions; +using System.Reflection; +using Newtonsoft.Json; + +namespace Firebase.Database.Offline.Internals +{ + public class MemberAccessVisitor : ExpressionVisitor + { + private readonly IList<string> propertyNames = new List<string>(); + + private bool wasDictionaryAccess; + + public IEnumerable<string> PropertyNames => propertyNames; + + public override Expression Visit(Expression expr) + { + if (expr?.NodeType == ExpressionType.MemberAccess) + { + if (wasDictionaryAccess) + { + wasDictionaryAccess = false; + } + else + { + var memberExpr = (MemberExpression) expr; + var jsonAttr = memberExpr.Member.GetCustomAttribute<JsonPropertyAttribute>(); + + propertyNames.Add(jsonAttr?.PropertyName ?? memberExpr.Member.Name); + } + } + else if (expr?.NodeType == ExpressionType.Call) + { + var callExpr = (MethodCallExpression) expr; + if (callExpr.Method.Name == "get_Item" && callExpr.Arguments.Count == 1) + { + var e = Expression.Lambda(callExpr.Arguments[0]).Compile(); + propertyNames.Add(e.DynamicInvoke().ToString()); + wasDictionaryAccess = callExpr.Arguments[0].NodeType == ExpressionType.MemberAccess; + } + } + + return base.Visit(expr); + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/OfflineCacheAdapter.cs b/dsa/FireBase/Offline/OfflineCacheAdapter.cs new file mode 100644 index 0000000..3153d1b --- /dev/null +++ b/dsa/FireBase/Offline/OfflineCacheAdapter.cs @@ -0,0 +1,152 @@ +using System; +using System.Collections; +using System.Collections.Generic; +using System.Linq; + +namespace Firebase.Database.Offline +{ + internal class OfflineCacheAdapter<TKey, T> : IDictionary<string, T>, IDictionary + { + private readonly IDictionary<string, OfflineEntry> database; + + public OfflineCacheAdapter(IDictionary<string, OfflineEntry> database) + { + this.database = database; + } + + public void CopyTo(Array array, int index) + { + throw new NotImplementedException(); + } + + public bool IsSynchronized { get; } + + public object SyncRoot { get; } + + object IDictionary.this[object key] + { + get => database[key.ToString()].Deserialize<T>(); + + set + { + var keyString = key.ToString(); + if (database.ContainsKey(keyString)) + database[keyString] = new OfflineEntry(keyString, value, database[keyString].Priority, + database[keyString].SyncOptions); + else + database[keyString] = new OfflineEntry(keyString, value, 1, SyncOptions.None); + } + } + + ICollection IDictionary.Values { get; } + + ICollection IDictionary.Keys { get; } + + public bool Contains(object key) + { + return ContainsKey(key.ToString()); + } + + IDictionaryEnumerator IDictionary.GetEnumerator() + { + throw new NotImplementedException(); + } + + public void Remove(object key) + { + Remove(key.ToString()); + } + + public bool IsFixedSize => false; + + public void Add(object key, object value) + { + Add(key.ToString(), (T) value); + } + + public int Count => database.Count; + + public bool IsReadOnly => database.IsReadOnly; + + public ICollection<string> Keys => database.Keys; + + public ICollection<T> Values => database.Values.Select(o => o.Deserialize<T>()).ToList(); + + public T this[string key] + { + get => database[key].Deserialize<T>(); + + set + { + if (database.ContainsKey(key)) + database[key] = new OfflineEntry(key, value, database[key].Priority, database[key].SyncOptions); + else + database[key] = new OfflineEntry(key, value, 1, SyncOptions.None); + } + } + + public IEnumerator<KeyValuePair<string, T>> GetEnumerator() + { + return database.Select(d => new KeyValuePair<string, T>(d.Key, d.Value.Deserialize<T>())).GetEnumerator(); + } + + IEnumerator IEnumerable.GetEnumerator() + { + return GetEnumerator(); + } + + public void Add(KeyValuePair<string, T> item) + { + Add(item.Key, item.Value); + } + + public void Clear() + { + database.Clear(); + } + + public bool Contains(KeyValuePair<string, T> item) + { + return ContainsKey(item.Key); + } + + public void CopyTo(KeyValuePair<string, T>[] array, int arrayIndex) + { + throw new NotImplementedException(); + } + + public bool Remove(KeyValuePair<string, T> item) + { + return database.Remove(item.Key); + } + + public void Add(string key, T value) + { + database.Add(key, new OfflineEntry(key, value, 1, SyncOptions.None)); + } + + public bool ContainsKey(string key) + { + return database.ContainsKey(key); + } + + public bool Remove(string key) + { + return database.Remove(key); + } + + public bool TryGetValue(string key, out T value) + { + OfflineEntry val; + + if (database.TryGetValue(key, out val)) + { + value = val.Deserialize<T>(); + return true; + } + + value = default(T); + return false; + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/OfflineDatabase.cs b/dsa/FireBase/Offline/OfflineDatabase.cs new file mode 100644 index 0000000..be0380b --- /dev/null +++ b/dsa/FireBase/Offline/OfflineDatabase.cs @@ -0,0 +1,228 @@ +using System; +using System.Collections; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using LiteDB; + +namespace Firebase.Database.Offline +{ + /// <summary> + /// The offline database. + /// </summary> + public class OfflineDatabase : IDictionary<string, OfflineEntry> + { + private readonly IDictionary<string, OfflineEntry> cache; + private readonly LiteRepository db; + + /// <summary> + /// Initializes a new instance of the <see cref="OfflineDatabase" /> class. + /// </summary> + /// <param name="itemType"> The item type which is used to determine the database file name. </param> + /// <param name="filenameModifier"> Custom string which will get appended to the file name. </param> + public OfflineDatabase(Type itemType, string filenameModifier) + { + var fullName = GetFileName(itemType.ToString()); + if (fullName.Length > 100) fullName = fullName.Substring(0, 100); + + var mapper = BsonMapper.Global; + mapper.Entity<OfflineEntry>().Id(o => o.Key); + + var root = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData); + var filename = fullName + filenameModifier + ".db"; + var path = Path.Combine(root, filename); + db = new LiteRepository(new LiteDatabase(path, mapper)); + + cache = db.Database.GetCollection<OfflineEntry>().FindAll() + .ToDictionary(o => o.Key, o => o); + } + + /// <summary> + /// Gets the number of elements contained in the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <returns> The number of elements contained in the <see cref="T:System.Collections.Generic.ICollection`1" />. </returns> + public int Count => cache.Count; + + /// <summary> + /// Gets a value indicating whether this is a read-only collection. + /// </summary> + public bool IsReadOnly => cache.IsReadOnly; + + /// <summary> + /// Gets an <see cref="T:System.Collections.Generic.ICollection`1" /> containing the keys of the + /// <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <returns> + /// An <see cref="T:System.Collections.Generic.ICollection`1" /> containing the keys of the object that + /// implements <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public ICollection<string> Keys => cache.Keys; + + /// <summary> + /// Gets an <see cref="T:System.Collections.Generic.ICollection`1" /> containing the values in the + /// <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <returns> + /// An <see cref="T:System.Collections.Generic.ICollection`1" /> containing the values in the object that + /// implements <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public ICollection<OfflineEntry> Values => cache.Values; + + /// <summary> + /// Gets or sets the element with the specified key. + /// </summary> + /// <param name="key">The key of the element to get or set.</param> + /// <returns> The element with the specified key. </returns> + public OfflineEntry this[string key] + { + get => cache[key]; + + set + { + cache[key] = value; + db.Upsert(value); + } + } + + /// <summary> + /// Returns an enumerator that iterates through the collection. + /// </summary> + /// <returns> An enumerator that can be used to iterate through the collection. </returns> + public IEnumerator<KeyValuePair<string, OfflineEntry>> GetEnumerator() + { + return cache.GetEnumerator(); + } + + IEnumerator IEnumerable.GetEnumerator() + { + return GetEnumerator(); + } + + /// <summary> + /// Adds an item to the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <param name="item">The object to add to the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + public void Add(KeyValuePair<string, OfflineEntry> item) + { + Add(item.Key, item.Value); + } + + /// <summary> + /// Removes all items from the <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + public void Clear() + { + cache.Clear(); + db.Delete<OfflineEntry>(LiteDB.Query.All()); + } + + /// <summary> + /// Determines whether the <see cref="T:System.Collections.Generic.ICollection`1" /> contains a specific value. + /// </summary> + /// <param name="item">The object to locate in the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + /// <returns> + /// True if <paramref name="item" /> is found in the <see cref="T:System.Collections.Generic.ICollection`1" />; + /// otherwise, false. + /// </returns> + public bool Contains(KeyValuePair<string, OfflineEntry> item) + { + return ContainsKey(item.Key); + } + + /// <summary> + /// Copies the elements of the <see cref="T:System.Collections.Generic.ICollection`1" /> to an + /// <see cref="T:System.Array" />, starting at a particular <see cref="T:System.Array" /> index. + /// </summary> + /// <param name="array"> + /// The one-dimensional <see cref="T:System.Array" /> that is the destination of the elements copied + /// from <see cref="T:System.Collections.Generic.ICollection`1" />. The <see cref="T:System.Array" /> must have + /// zero-based indexing. + /// </param> + /// <param name="arrayIndex">The zero-based index in <paramref name="array" /> at which copying begins.</param> + public void CopyTo(KeyValuePair<string, OfflineEntry>[] array, int arrayIndex) + { + cache.CopyTo(array, arrayIndex); + } + + /// <summary> + /// Removes the first occurrence of a specific object from the + /// <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </summary> + /// <param name="item">The object to remove from the <see cref="T:System.Collections.Generic.ICollection`1" />.</param> + /// <returns> + /// True if <paramref name="item" /> was successfully removed from the + /// <see cref="T:System.Collections.Generic.ICollection`1" />; otherwise, false. This method also returns false if + /// <paramref name="item" /> is not found in the original <see cref="T:System.Collections.Generic.ICollection`1" />. + /// </returns> + public bool Remove(KeyValuePair<string, OfflineEntry> item) + { + return Remove(item.Key); + } + + /// <summary> + /// Determines whether the <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an element with the + /// specified key. + /// </summary> + /// <param name="key">The key to locate in the <see cref="T:System.Collections.Generic.IDictionary`2" />.</param> + /// <returns> + /// True if the <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an element with the key; + /// otherwise, false. + /// </returns> + public bool ContainsKey(string key) + { + return cache.ContainsKey(key); + } + + /// <summary> + /// Adds an element with the provided key and value to the <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <param name="key">The object to use as the key of the element to add.</param> + /// <param name="value">The object to use as the value of the element to add.</param> + public void Add(string key, OfflineEntry value) + { + cache.Add(key, value); + db.Insert(value); + } + + /// <summary> + /// Removes the element with the specified key from the <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </summary> + /// <param name="key">The key of the element to remove.</param> + /// <returns> + /// True if the element is successfully removed; otherwise, false. This method also returns false if + /// <paramref name="key" /> was not found in the original <see cref="T:System.Collections.Generic.IDictionary`2" />. + /// </returns> + public bool Remove(string key) + { + cache.Remove(key); + return db.Delete<OfflineEntry>(key); + } + + /// <summary> + /// Gets the value associated with the specified key. + /// </summary> + /// <param name="key">The key whose value to get.</param> + /// <param name="value"> + /// When this method returns, the value associated with the specified key, if the key is found; + /// otherwise, the default value for the type of the <paramref name="value" /> parameter. This parameter is passed + /// uninitialized. + /// </param> + /// <returns> + /// True if the object that implements <see cref="T:System.Collections.Generic.IDictionary`2" /> contains an + /// element with the specified key; otherwise, false. + /// </returns> + public bool TryGetValue(string key, out OfflineEntry value) + { + return cache.TryGetValue(key, out value); + } + + private string GetFileName(string fileName) + { + var invalidChars = new[] {'`', '[', ',', '='}; + foreach (var c in invalidChars.Concat(Path.GetInvalidFileNameChars()).Distinct()) + fileName = fileName.Replace(c, '_'); + + return fileName; + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/OfflineEntry.cs b/dsa/FireBase/Offline/OfflineEntry.cs new file mode 100644 index 0000000..9feffa3 --- /dev/null +++ b/dsa/FireBase/Offline/OfflineEntry.cs @@ -0,0 +1,99 @@ +using System; +using Newtonsoft.Json; + +namespace Firebase.Database.Offline +{ + /// <summary> + /// Represents an object stored in offline storage. + /// </summary> + public class OfflineEntry + { + private object dataInstance; + + /// <summary> + /// Initializes a new instance of the <see cref="OfflineEntry" /> class with an already serialized object. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="obj"> The object. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + /// <param name="syncOptions"> The sync options. </param> + public OfflineEntry(string key, object obj, string data, int priority, SyncOptions syncOptions, + bool isPartial = false) + { + Key = key; + Priority = priority; + Data = data; + Timestamp = DateTime.UtcNow; + SyncOptions = syncOptions; + IsPartial = isPartial; + + dataInstance = obj; + } + + /// <summary> + /// Initializes a new instance of the <see cref="OfflineEntry" /> class. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="obj"> The object. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + /// <param name="syncOptions"> The sync options. </param> + public OfflineEntry(string key, object obj, int priority, SyncOptions syncOptions, bool isPartial = false) + : this(key, obj, JsonConvert.SerializeObject(obj), priority, syncOptions, isPartial) + { + } + + /// <summary> + /// Initializes a new instance of the <see cref="OfflineEntry" /> class. + /// </summary> + public OfflineEntry() + { + } + + /// <summary> + /// Gets or sets the key of this entry. + /// </summary> + public string Key { get; set; } + + /// <summary> + /// Gets or sets the priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </summary> + public int Priority { get; set; } + + /// <summary> + /// Gets or sets the timestamp when this entry was last touched. + /// </summary> + public DateTime Timestamp { get; set; } + + /// <summary> + /// Gets or sets the <see cref="SyncOptions" /> which define what sync state this entry is in. + /// </summary> + public SyncOptions SyncOptions { get; set; } + + /// <summary> + /// Gets or sets serialized JSON data. + /// </summary> + public string Data { get; set; } + + /// <summary> + /// Specifies whether this is only a partial object. + /// </summary> + public bool IsPartial { get; set; } + + /// <summary> + /// Deserializes <see cref="Data" /> into <typeparamref name="T" />. The result is cached. + /// </summary> + /// <typeparam name="T"> Type of object to deserialize into. </typeparam> + /// <returns> Instance of <typeparamref name="T" />. </returns> + public T Deserialize<T>() + { + return (T) (dataInstance ?? (dataInstance = JsonConvert.DeserializeObject<T>(Data))); + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/RealtimeDatabase.cs b/dsa/FireBase/Offline/RealtimeDatabase.cs new file mode 100644 index 0000000..973db46 --- /dev/null +++ b/dsa/FireBase/Offline/RealtimeDatabase.cs @@ -0,0 +1,479 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Linq.Expressions; +using System.Net; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reactive.Threading.Tasks; +using System.Reflection; +using System.Threading; +using System.Threading.Tasks; +using Firebase.Database.Extensions; +using Firebase.Database.Offline.Internals; +using Firebase.Database.Query; +using Firebase.Database.Streaming; +using Newtonsoft.Json; + +namespace Firebase.Database.Offline +{ + /// <summary> + /// The real-time Database which synchronizes online and offline data. + /// </summary> + /// <typeparam name="T"> Type of entities. </typeparam> + public class RealtimeDatabase<T> : IDisposable where T : class + { + private readonly ChildQuery childQuery; + private readonly string elementRoot; + private readonly FirebaseCache<T> firebaseCache; + private readonly InitialPullStrategy initialPullStrategy; + private readonly bool pushChanges; + private readonly StreamingOptions streamingOptions; + private readonly Subject<FirebaseEvent<T>> subject; + private FirebaseSubscription<T> firebaseSubscription; + + private bool isSyncRunning; + private IObservable<FirebaseEvent<T>> observable; + + /// <summary> + /// Initializes a new instance of the <see cref="RealtimeDatabase{T}" /> class. + /// </summary> + /// <param name="childQuery"> The child query. </param> + /// <param name="elementRoot"> The element Root. </param> + /// <param name="offlineDatabaseFactory"> The offline database factory. </param> + /// <param name="filenameModifier"> Custom string which will get appended to the file name. </param> + /// <param name="streamChanges"> Specifies whether changes should be streamed from the server. </param> + /// <param name="pullEverythingOnStart"> + /// Specifies if everything should be pull from the online storage on start. It only + /// makes sense when <see cref="streamChanges" /> is set to true. + /// </param> + /// <param name="pushChanges"> + /// Specifies whether changed items should actually be pushed to the server. If this is false, + /// then Put / Post / Delete will not affect server data. + /// </param> + public RealtimeDatabase(ChildQuery childQuery, string elementRoot, + Func<Type, string, IDictionary<string, OfflineEntry>> offlineDatabaseFactory, string filenameModifier, + StreamingOptions streamingOptions, InitialPullStrategy initialPullStrategy, bool pushChanges, + ISetHandler<T> setHandler = null) + { + this.childQuery = childQuery; + this.elementRoot = elementRoot; + this.streamingOptions = streamingOptions; + this.initialPullStrategy = initialPullStrategy; + this.pushChanges = pushChanges; + Database = offlineDatabaseFactory(typeof(T), filenameModifier); + firebaseCache = new FirebaseCache<T>(new OfflineCacheAdapter<string, T>(Database)); + subject = new Subject<FirebaseEvent<T>>(); + + PutHandler = setHandler ?? new SetHandler<T>(); + + isSyncRunning = true; + Task.Factory.StartNew(SynchronizeThread, CancellationToken.None, TaskCreationOptions.LongRunning, + TaskScheduler.Default); + } + + /// <summary> + /// Gets the backing Database. + /// </summary> + public IDictionary<string, OfflineEntry> Database { get; } + + public ISetHandler<T> PutHandler { private get; set; } + + public void Dispose() + { + subject.OnCompleted(); + firebaseSubscription?.Dispose(); + } + + /// <summary> + /// Event raised whenever an exception is thrown in the synchronization thread. Exception thrown in there are + /// swallowed, so this event is the only way to get to them. + /// </summary> + public event EventHandler<ExceptionEventArgs> SyncExceptionThrown; + + /// <summary> + /// Overwrites existing object with given key. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="obj"> The object to set. </param> + /// <param name="syncOnline"> Indicates whether the item should be synced online. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public void Set(string key, T obj, SyncOptions syncOptions, int priority = 1) + { + SetAndRaise(key, new OfflineEntry(key, obj, priority, syncOptions)); + } + + public void Set<TProperty>(string key, Expression<Func<T, TProperty>> propertyExpression, object value, + SyncOptions syncOptions, int priority = 1) + { + var fullKey = GenerateFullKey(key, propertyExpression, syncOptions); + var serializedObject = JsonConvert.SerializeObject(value).Trim('"', '\\'); + + if (fullKey.Item3) + { + if (typeof(TProperty) != typeof(string) || value == null) + // don't escape non-string primitives and null; + serializedObject = $"{{ \"{fullKey.Item2}\" : {serializedObject} }}"; + else + serializedObject = $"{{ \"{fullKey.Item2}\" : \"{serializedObject}\" }}"; + } + + var setObject = firebaseCache.PushData("/" + fullKey.Item1, serializedObject).First(); + + if (!Database.ContainsKey(key) || Database[key].SyncOptions != SyncOptions.Patch && + Database[key].SyncOptions != SyncOptions.Put) + Database[fullKey.Item1] = + new OfflineEntry(fullKey.Item1, value, serializedObject, priority, syncOptions, true); + + subject.OnNext(new FirebaseEvent<T>(key, setObject.Object, + setObject == null ? FirebaseEventType.Delete : FirebaseEventType.InsertOrUpdate, + FirebaseEventSource.Offline)); + } + + /// <summary> + /// Fetches an object with the given key and adds it to the Database. + /// </summary> + /// <param name="key"> The key. </param> + /// <param name="priority"> + /// The priority. Objects with higher priority will be synced first. Higher number indicates higher + /// priority. + /// </param> + public void Pull(string key, int priority = 1) + { + if (!Database.ContainsKey(key)) + Database[key] = new OfflineEntry(key, null, priority, SyncOptions.Pull); + else if (Database[key].SyncOptions == SyncOptions.None) + // pull only if push isn't pending + Database[key].SyncOptions = SyncOptions.Pull; + } + + /// <summary> + /// Fetches everything from the remote database. + /// </summary> + public async Task PullAsync() + { + var existingEntries = await childQuery + .OnceAsync<T>() + .ToObservable() + .RetryAfterDelay<IReadOnlyCollection<FirebaseObject<T>>, FirebaseException>( + childQuery.Client.Options.SyncPeriod, + ex => ex.StatusCode == + HttpStatusCode + .OK) // OK implies the request couldn't complete due to network error. + .Select(e => ResetDatabaseFromInitial(e, false)) + .SelectMany(e => e) + .Do(e => + { + Database[e.Key] = new OfflineEntry(e.Key, e.Object, 1, SyncOptions.None); + subject.OnNext(new FirebaseEvent<T>(e.Key, e.Object, FirebaseEventType.InsertOrUpdate, + FirebaseEventSource.OnlinePull)); + }) + .ToList(); + + // Remove items not stored online + foreach (var item in Database.Keys.Except(existingEntries.Select(f => f.Key)).ToList()) + { + Database.Remove(item); + subject.OnNext(new FirebaseEvent<T>(item, null, FirebaseEventType.Delete, + FirebaseEventSource.OnlinePull)); + } + } + + /// <summary> + /// Retrieves all offline items currently stored in local database. + /// </summary> + public IEnumerable<FirebaseObject<T>> Once() + { + return Database + .Where(kvp => !string.IsNullOrEmpty(kvp.Value.Data) && kvp.Value.Data != "null" && !kvp.Value.IsPartial) + .Select(kvp => new FirebaseObject<T>(kvp.Key, kvp.Value.Deserialize<T>())) + .ToList(); + } + + /// <summary> + /// Starts observing the real-time Database. Events will be fired both when change is done locally and remotely. + /// </summary> + /// <returns> Stream of <see cref="FirebaseEvent{T}" />. </returns> + public IObservable<FirebaseEvent<T>> AsObservable() + { + if (!isSyncRunning) + { + isSyncRunning = true; + Task.Factory.StartNew(SynchronizeThread, CancellationToken.None, TaskCreationOptions.LongRunning, + TaskScheduler.Default); + } + + if (observable == null) + { + var initialData = Observable.Return(FirebaseEvent<T>.Empty(FirebaseEventSource.Offline)); + if (Database.TryGetValue(elementRoot, out var oe)) + initialData = Observable.Return(oe) + .Where(offlineEntry => + !string.IsNullOrEmpty(offlineEntry.Data) && offlineEntry.Data != "null" && + !offlineEntry.IsPartial) + .Select(offlineEntry => new FirebaseEvent<T>(offlineEntry.Key, offlineEntry.Deserialize<T>(), + FirebaseEventType.InsertOrUpdate, FirebaseEventSource.Offline)); + else if (Database.Count > 0) + initialData = Database + .Where(kvp => + !string.IsNullOrEmpty(kvp.Value.Data) && kvp.Value.Data != "null" && !kvp.Value.IsPartial) + .Select(kvp => new FirebaseEvent<T>(kvp.Key, kvp.Value.Deserialize<T>(), + FirebaseEventType.InsertOrUpdate, FirebaseEventSource.Offline)) + .ToList() + .ToObservable(); + + observable = initialData + .Merge(subject) + .Merge(GetInitialPullObservable() + .RetryAfterDelay<IReadOnlyCollection<FirebaseObject<T>>, FirebaseException>( + childQuery.Client.Options.SyncPeriod, + ex => ex.StatusCode == + HttpStatusCode + .OK) // OK implies the request couldn't complete due to network error. + .Select(e => ResetDatabaseFromInitial(e)) + .SelectMany(e => e) + .Do(SetObjectFromInitialPull) + .Select(e => new FirebaseEvent<T>(e.Key, e.Object, + e.Object == null ? FirebaseEventType.Delete : FirebaseEventType.InsertOrUpdate, + FirebaseEventSource.OnlineInitial)) + .Concat(Observable.Create<FirebaseEvent<T>>(observer => + InitializeStreamingSubscription(observer)))) + .Do(next => { }, e => observable = null, () => observable = null) + .Replay() + .RefCount(); + } + + return observable; + } + + private IReadOnlyCollection<FirebaseObject<T>> ResetDatabaseFromInitial( + IReadOnlyCollection<FirebaseObject<T>> collection, bool onlyWhenInitialEverything = true) + { + if (onlyWhenInitialEverything && initialPullStrategy != InitialPullStrategy.Everything) return collection; + + // items which are in local db, but not in the online collection + var extra = Once() + .Select(f => f.Key) + .Except(collection.Select(c => c.Key)) + .Select(k => new FirebaseObject<T>(k, null)); + + return collection.Concat(extra).ToList(); + } + + private void SetObjectFromInitialPull(FirebaseObject<T> e) + { + // set object with no sync only if it doesn't exist yet + // and the InitialPullStrategy != Everything + // this attempts to deal with scenario when you are offline, have local changes and go online + // in this case having the InitialPullStrategy set to everything would basically purge all local changes + if (!Database.ContainsKey(e.Key) || Database[e.Key].SyncOptions == SyncOptions.None || + Database[e.Key].SyncOptions == SyncOptions.Pull || + initialPullStrategy != InitialPullStrategy.Everything) + Database[e.Key] = new OfflineEntry(e.Key, e.Object, 1, SyncOptions.None); + } + + private IObservable<IReadOnlyCollection<FirebaseObject<T>>> GetInitialPullObservable() + { + FirebaseQuery query; + switch (initialPullStrategy) + { + case InitialPullStrategy.MissingOnly: + query = childQuery.OrderByKey().StartAt(() => GetLatestKey()); + break; + case InitialPullStrategy.Everything: + query = childQuery; + break; + case InitialPullStrategy.None: + default: + return Observable.Empty<IReadOnlyCollection<FirebaseEvent<T>>>(); + } + + if (string.IsNullOrWhiteSpace(elementRoot)) + return Observable.Defer(() => query.OnceAsync<T>().ToObservable()); + + // there is an element root, which indicates the target location is not a collection but a single element + return Observable.Defer(async () => + Observable.Return(await query.OnceSingleAsync<T>()) + .Select(e => new[] {new FirebaseObject<T>(elementRoot, e)})); + } + + private IDisposable InitializeStreamingSubscription(IObserver<FirebaseEvent<T>> observer) + { + var completeDisposable = Disposable.Create(() => isSyncRunning = false); + + switch (streamingOptions) + { + case StreamingOptions.LatestOnly: + // stream since the latest key + var queryLatest = childQuery.OrderByKey().StartAt(() => GetLatestKey()); + firebaseSubscription = + new FirebaseSubscription<T>(observer, queryLatest, elementRoot, firebaseCache); + firebaseSubscription.ExceptionThrown += StreamingExceptionThrown; + + return new CompositeDisposable(firebaseSubscription.Run(), completeDisposable); + case StreamingOptions.Everything: + // stream everything + var queryAll = childQuery; + firebaseSubscription = new FirebaseSubscription<T>(observer, queryAll, elementRoot, firebaseCache); + firebaseSubscription.ExceptionThrown += StreamingExceptionThrown; + + return new CompositeDisposable(firebaseSubscription.Run(), completeDisposable); + } + + return completeDisposable; + } + + private void SetAndRaise(string key, OfflineEntry obj, + FirebaseEventSource eventSource = FirebaseEventSource.Offline) + { + Database[key] = obj; + subject.OnNext(new FirebaseEvent<T>(key, obj?.Deserialize<T>(), + string.IsNullOrEmpty(obj?.Data) || obj?.Data == "null" + ? FirebaseEventType.Delete + : FirebaseEventType.InsertOrUpdate, eventSource)); + } + + private async void SynchronizeThread() + { + while (isSyncRunning) + { + try + { + var validEntries = Database.Where(e => e.Value != null); + await PullEntriesAsync(validEntries.Where(kvp => kvp.Value.SyncOptions == SyncOptions.Pull)); + + if (pushChanges) + await PushEntriesAsync(validEntries.Where(kvp => + kvp.Value.SyncOptions == SyncOptions.Put || kvp.Value.SyncOptions == SyncOptions.Patch)); + } + catch (Exception ex) + { + SyncExceptionThrown?.Invoke(this, new ExceptionEventArgs(ex)); + } + + await Task.Delay(childQuery.Client.Options.SyncPeriod); + } + } + + private string GetLatestKey() + { + var key = Database.OrderBy(o => o.Key, StringComparer.Ordinal).LastOrDefault().Key ?? string.Empty; + + if (!string.IsNullOrWhiteSpace(key)) + key = key.Substring(0, key.Length - 1) + (char) (key[key.Length - 1] + 1); + + return key; + } + + private async Task PushEntriesAsync(IEnumerable<KeyValuePair<string, OfflineEntry>> pushEntries) + { + var groups = pushEntries.GroupBy(pair => pair.Value.Priority).OrderByDescending(kvp => kvp.Key).ToList(); + + foreach (var group in groups) + { + var tasks = group.OrderBy(kvp => kvp.Value.IsPartial).Select(kvp => + kvp.Value.IsPartial + ? ResetSyncAfterPush(PutHandler.SetAsync(childQuery, kvp.Key, kvp.Value), kvp.Key) + : ResetSyncAfterPush(PutHandler.SetAsync(childQuery, kvp.Key, kvp.Value), kvp.Key, + kvp.Value.Deserialize<T>())); + + try + { + await Task.WhenAll(tasks).WithAggregateException(); + } + catch (Exception ex) + { + SyncExceptionThrown?.Invoke(this, new ExceptionEventArgs(ex)); + } + } + } + + private async Task PullEntriesAsync(IEnumerable<KeyValuePair<string, OfflineEntry>> pullEntries) + { + var taskGroups = pullEntries.GroupBy(pair => pair.Value.Priority).OrderByDescending(kvp => kvp.Key); + + foreach (var group in taskGroups) + { + var tasks = group.Select(pair => + ResetAfterPull( + childQuery.Child(pair.Key == elementRoot ? string.Empty : pair.Key).OnceSingleAsync<T>(), + pair.Key, pair.Value)); + + try + { + await Task.WhenAll(tasks).WithAggregateException(); + } + catch (Exception ex) + { + SyncExceptionThrown?.Invoke(this, new ExceptionEventArgs(ex)); + } + } + } + + private async Task ResetAfterPull(Task<T> task, string key, OfflineEntry entry) + { + await task; + SetAndRaise(key, new OfflineEntry(key, task.Result, entry.Priority, SyncOptions.None), + FirebaseEventSource.OnlinePull); + } + + private async Task ResetSyncAfterPush(Task task, string key, T obj) + { + await ResetSyncAfterPush(task, key); + + if (streamingOptions == StreamingOptions.None) + subject.OnNext(new FirebaseEvent<T>(key, obj, + obj == null ? FirebaseEventType.Delete : FirebaseEventType.InsertOrUpdate, + FirebaseEventSource.OnlinePush)); + } + + private async Task ResetSyncAfterPush(Task task, string key) + { + await task; + ResetSyncOptions(key); + } + + private void ResetSyncOptions(string key) + { + var item = Database[key]; + + if (item.IsPartial) + { + Database.Remove(key); + } + else + { + item.SyncOptions = SyncOptions.None; + Database[key] = item; + } + } + + private void StreamingExceptionThrown(object sender, ExceptionEventArgs<FirebaseException> e) + { + SyncExceptionThrown?.Invoke(this, new ExceptionEventArgs(e.Exception)); + } + + private Tuple<string, string, bool> GenerateFullKey<TProperty>(string key, + Expression<Func<T, TProperty>> propertyGetter, SyncOptions syncOptions) + { + var visitor = new MemberAccessVisitor(); + visitor.Visit(propertyGetter); + var propertyType = typeof(TProperty).GetTypeInfo(); + var prefix = key == string.Empty ? string.Empty : key + "/"; + + // primitive types + if (syncOptions == SyncOptions.Patch && (propertyType.IsPrimitive || + Nullable.GetUnderlyingType(typeof(TProperty)) != null || + typeof(TProperty) == typeof(string))) + return Tuple.Create(prefix + string.Join("/", visitor.PropertyNames.Skip(1).Reverse()), + visitor.PropertyNames.First(), true); + + return Tuple.Create(prefix + string.Join("/", visitor.PropertyNames.Reverse()), + visitor.PropertyNames.First(), false); + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/SetHandler.cs b/dsa/FireBase/Offline/SetHandler.cs new file mode 100644 index 0000000..6314c3c --- /dev/null +++ b/dsa/FireBase/Offline/SetHandler.cs @@ -0,0 +1,19 @@ +using System.Threading.Tasks; +using Firebase.Database.Query; + +namespace Firebase.Database.Offline +{ + public class SetHandler<T> : ISetHandler<T> + { + public virtual async Task SetAsync(ChildQuery query, string key, OfflineEntry entry) + { + using (var child = query.Child(key)) + { + if (entry.SyncOptions == SyncOptions.Put) + await child.PutAsync(entry.Data); + else + await child.PatchAsync(entry.Data); + } + } + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/StreamingOptions.cs b/dsa/FireBase/Offline/StreamingOptions.cs new file mode 100644 index 0000000..a420cbb --- /dev/null +++ b/dsa/FireBase/Offline/StreamingOptions.cs @@ -0,0 +1,23 @@ +namespace Firebase.Database.Offline +{ + public enum StreamingOptions + { + /// <summary> + /// No realtime streaming. + /// </summary> + None, + + /// <summary> + /// Streaming of only new items - not the existing ones. + /// </summary> + LatestOnly, + + /// <summary> + /// Streaming of all items. This will also pull all existing items on start, so be mindful about the number of items in + /// your DB. + /// When used, consider not setting the <see cref="InitialPullStrategy" /> to + /// <see cref="InitialPullStrategy.Everything" /> because you would pointlessly pull everything twice. + /// </summary> + Everything + } +}
\ No newline at end of file diff --git a/dsa/FireBase/Offline/SyncOptions.cs b/dsa/FireBase/Offline/SyncOptions.cs new file mode 100644 index 0000000..ca68d0a --- /dev/null +++ b/dsa/FireBase/Offline/SyncOptions.cs @@ -0,0 +1,28 @@ +namespace Firebase.Database.Offline +{ + /// <summary> + /// Specifies type of sync requested for given data. + /// </summary> + public enum SyncOptions + { + /// <summary> + /// No sync needed for given data. + /// </summary> + None, + + /// <summary> + /// Data should be pulled from firebase. + /// </summary> + Pull, + + /// <summary> + /// Data should be put to firebase. + /// </summary> + Put, + + /// <summary> + /// Data should be patched in firebase. + /// </summary> + Patch + } +}
\ No newline at end of file |