using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Reactive; using System.Reactive.Subjects; using System.Data.Entity.Infrastructure; using System.Collections.Concurrent; using System.Data.Objects; namespace Disco.Data.Repository.Monitor { public static class RepositoryMonitor { private static Subject streamBefore = new Subject(); private static Subject streamAfter = new Subject(); private static ConcurrentDictionary entityProxyTypeCache = new ConcurrentDictionary(); public static Subject StreamBeforeCommit { get { return streamBefore; } } public static Subject StreamAfterCommit { get { return streamAfter; } } internal static RepositoryMonitorEvent[] BeforeSaveChanges(DiscoDataContext dbContext) { var contextStateManager = ((IObjectContextAdapter)dbContext).ObjectContext.ObjectStateManager; dbContext.ChangeTracker.DetectChanges(); var changes = dbContext.ChangeTracker.Entries().Where(entry => entry.State == System.Data.EntityState.Added || entry.State == System.Data.EntityState.Deleted || entry.State == System.Data.EntityState.Modified); var events = changes.Select(entryState => { ObjectStateEntry stateEntry = contextStateManager.GetObjectStateEntry(entryState.Entity); var monitorEvent = EventFromEntryState(dbContext, entryState, stateEntry); // Push to Stream streamBefore.OnNext(monitorEvent); return monitorEvent; }).ToArray(); return events; } internal static void AfterSaveChanges(DiscoDataContext dbContext, IEnumerable changes) { foreach (var change in changes) { UpdateAfterEventFromEntryState(change); streamAfter.OnNext(change); } } private static Type EntityTypeFromProxy(Type EntityProxyType) { Type EntityType; if (entityProxyTypeCache.TryGetValue(EntityProxyType, out EntityType)) return EntityType; EntityType = EntityProxyType; do { if (EntityType.Namespace.StartsWith("Disco.Models.Repository")) { entityProxyTypeCache.TryAdd(EntityProxyType, EntityType); return EntityType; } EntityType = EntityType.BaseType; } while (EntityType != null); throw new ArgumentException("The EntryProxyType does not inherit from any Repository Models", "EntityProxyType"); } internal static void UpdateAfterEventFromEntryState(RepositoryMonitorEvent monitorEvent) { if (monitorEvent.EventType == RepositoryMonitorEventType.Added) { // Update Entity Key for Added Events monitorEvent.EntityKey = monitorEvent.objectEntryState.EntityKey.EntityKeyValues.Select(kv => kv.Value).ToArray(); } } internal static RepositoryMonitorEvent EventFromEntryState(DiscoDataContext dbContext, DbEntityEntry entityEntry, ObjectStateEntry entryState) { RepositoryMonitorEventType eventType; string[] modifiedProperties = null; object[] entityKey = null; Type entityType; switch (entryState.State) { case System.Data.EntityState.Added: eventType = RepositoryMonitorEventType.Added; break; case System.Data.EntityState.Deleted: eventType = RepositoryMonitorEventType.Deleted; break; case System.Data.EntityState.Detached: eventType = RepositoryMonitorEventType.Detached; break; case System.Data.EntityState.Modified: eventType = RepositoryMonitorEventType.Modified; break; case System.Data.EntityState.Unchanged: eventType = RepositoryMonitorEventType.Unchanged; break; default: throw new NotSupportedException(string.Format("Database Entry State not supported: {0}", entryState.State.ToString())); } entityType = EntityTypeFromProxy(entryState.Entity.GetType()); // Only pass modified properties on Modified Event if (eventType == RepositoryMonitorEventType.Modified) modifiedProperties = entryState.GetModifiedProperties().ToArray(); else modifiedProperties = new string[] { }; // Empty array for Added/Deleted. // Don't pass entity key when entity newly added if (eventType != RepositoryMonitorEventType.Added) entityKey = entryState.EntityKey.EntityKeyValues.Select(kv => kv.Value).ToArray(); return new RepositoryMonitorEvent() { EventType = eventType, Entity = entryState.Entity, EntityKey = entityKey, EntityType = entityType, ModifiedProperties = modifiedProperties, dbContext = dbContext, dbEntityState = entityEntry, objectEntryState = entryState }; } } }