Merge pull request danielgerlag#277 from knoxi/features/issue-261

danielgerlag · web-flow · commit c2d0aced36d2 · 2019-03-16T14:42:23.000-07:00
add support to get multiple workflow instances by a list of ids
diff --git a/src/WorkflowCore/Interface/IPersistenceProvider.cs b/src/WorkflowCore/Interface/IPersistenceProvider.cs
@@ -7,7 +7,7 @@ namespace WorkflowCore.Interface
 {
     /// <remarks>
     /// The implemention of this interface will be responsible for
-    /// persisiting running workflow instances to a durable store    
+    /// persisting running workflow instances to a durable store
     /// </remarks>
     public interface IPersistenceProvider
     {
@@ -22,6 +22,8 @@ public interface IPersistenceProvider
 
         Task<WorkflowInstance> GetWorkflowInstance(string Id);
 
+        Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids);
+
         Task<string> CreateEventSubscription(EventSubscription subscription);
 
         Task<IEnumerable<EventSubscription>> GetSubcriptions(string eventName, string eventKey, DateTime asOf);
diff --git a/src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs b/src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs
@@ -8,22 +8,22 @@
 
 namespace WorkflowCore.Services
 {
-    
+
     public interface ISingletonMemoryProvider : IPersistenceProvider
     {
     }
-    #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
 
     /// <summary>
     /// In-memory implementation of IPersistenceProvider for demo and testing purposes
     /// </summary>
     public class MemoryPersistenceProvider : ISingletonMemoryProvider
-    {        
+    {
         private readonly List<WorkflowInstance> _instances = new List<WorkflowInstance>();
         private readonly List<EventSubscription> _subscriptions = new List<EventSubscription>();
         private readonly List<Event> _events = new List<Event>();
         private readonly List<ExecutionError> _errors = new List<ExecutionError>();
-                
+
         public async Task<string> CreateNewWorkflow(WorkflowInstance workflow)
         {
             lock (_instances)
@@ -61,6 +61,19 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             }
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            lock (_instances)
+            {
+                return _instances.Where(x => ids.Contains(x.Id));
+            }
+        }
+
         public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
         {
             lock (_instances)
@@ -121,7 +134,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)
         }
 
         public void EnsureStoreExists()
-        {            
+        {
         }
 
         public async Task<string> CreateEvent(Event newEvent)
@@ -133,7 +146,7 @@ public async Task<string> CreateEvent(Event newEvent)
                 return newEvent.Id;
             }
         }
-        
+
         public async Task MarkEventProcessed(string id)
         {
             lock (_events)
@@ -197,5 +210,5 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)
         }
     }
 
-    #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
 }
diff --git a/src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs b/src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs
@@ -36,6 +36,8 @@ public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
 
         public Task<WorkflowInstance> GetWorkflowInstance(string Id) => _innerService.GetWorkflowInstance(Id);
 
+        public Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids) => _innerService.GetWorkflowInstances(ids);
+
         public Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take) => _innerService.GetWorkflowInstances(status, type, createdFrom, createdTo, skip, take);
 
         public Task MarkEventProcessed(string id) => _innerService.MarkEventProcessed(id);
diff --git a/src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.EntityFramework/Services/EntityFrameworkPersistenceProvider.cs
@@ -26,7 +26,7 @@ public EntityFrameworkPersistenceProvider(IWorkflowDbContextFactory contextFacto
             _canCreateDB = canCreateDB;
             _canMigrateDB = canMigrateDB;
         }
-        
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             using (var db = ConstructDbContext())
@@ -96,7 +96,7 @@ public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowSt
                 return result;
             }
         }
-        
+
         public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
         {
             using (var db = ConstructDbContext())
@@ -115,6 +115,26 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             }
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            using (var db = ConstructDbContext())
+            {
+                var uids = ids.Select(i => new Guid(i));
+                var raw = db.Set<PersistedWorkflow>()
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .Where(x => uids.Contains(x.InstanceId));
+
+                return (await raw.ToListAsync()).Select(i => i.ToWorkflowInstance());
+            }
+        }
+
         public async Task PersistWorkflow(WorkflowInstance workflow)
         {
             using (var db = ConstructDbContext())
@@ -143,7 +163,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)
                 await db.SaveChangesAsync();
             }
         }
-                
+
         public virtual void EnsureStoreExists()
         {
             using (var context = ConstructDbContext())
@@ -283,7 +303,7 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)
                 }
             }
         }
-        
+
         private WorkflowDbContext ConstructDbContext()
         {
             return _contextFactory.Build();
diff --git a/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs b/src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs
@@ -26,8 +26,8 @@ public MongoPersistenceProvider(IMongoDatabase database)
         static MongoPersistenceProvider()
         {
             BsonClassMap.RegisterClassMap<WorkflowInstance>(x =>
-            {                
-                x.MapIdProperty(y => y.Id)                    
+            {
+                x.MapIdProperty(y => y.Id)
                     .SetIdGenerator(new StringObjectIdGenerator());
                 x.MapProperty(y => y.Data)
                     .SetSerializer(new DataObjectSerializer());
@@ -48,9 +48,9 @@ static MongoPersistenceProvider()
                     .SetIdGenerator(new StringObjectIdGenerator());
                 x.MapProperty(y => y.EventName);
                 x.MapProperty(y => y.EventKey);
-                x.MapProperty(y => y.StepId);                
+                x.MapProperty(y => y.StepId);
                 x.MapProperty(y => y.WorkflowId);
-                x.MapProperty(y => y.SubscribeAsOf);                
+                x.MapProperty(y => y.SubscribeAsOf);
             });
 
             BsonClassMap.RegisterClassMap<Event>(x =>
@@ -113,7 +113,18 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             var result = await WorkflowInstances.FindAsync(x => x.Id == Id);
             return await result.FirstAsync();
         }
-        
+
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var result = await WorkflowInstances.FindAsync(x => ids.Contains(x.Id));
+            return await result.ToListAsync();
+        }
+
         public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
         {
             IQueryable<WorkflowInstance> result = WorkflowInstances.AsQueryable();
@@ -138,17 +149,17 @@ public async Task<string> CreateEventSubscription(EventSubscription subscription
             await EventSubscriptions.InsertOneAsync(subscription);
             return subscription.Id;
         }
-        
+
         public async Task TerminateSubscription(string eventSubscriptionId)
         {
             await EventSubscriptions.DeleteOneAsync(x => x.Id == eventSubscriptionId);
         }
 
         public void EnsureStoreExists()
         {
-            
-        }                
-                
+
+        }
+
         public async Task<IEnumerable<EventSubscription>> GetSubcriptions(string eventName, string eventKey, DateTime asOf)
         {
             var query = EventSubscriptions
@@ -192,7 +203,7 @@ public async Task<IEnumerable<string>> GetEvents(string eventName, string eventK
             var query = Events
                 .Find(x => x.EventName == eventName && x.EventKey == eventKey && x.EventTime >= asOf)
                 .Project(x => x.Id);
-            
+
             return await query.ToListAsync();
         }
 
diff --git a/src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs
@@ -4,6 +4,7 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using WorkflowCore.Providers.AWS.Interface;
@@ -86,7 +87,7 @@ public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt)
                 },
                 ScanIndexForward = true
             };
-            
+
             var response = await _client.QueryAsync(request);
 
             foreach (var item in response.Items)
@@ -117,6 +118,57 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             return response.Item.ToWorkflowInstance();
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var keys = new KeysAndAttributes() { Keys = new List<Dictionary<string, AttributeValue>>() };
+            foreach (var id in ids)
+            {
+                var key = new Dictionary<string, AttributeValue>()
+                {
+                    {
+                        "id", new AttributeValue { S = id }
+                    }
+                };
+                keys.Keys.Add(key);
+            }
+
+            var request = new BatchGetItemRequest
+            {
+                RequestItems = new Dictionary<string, KeysAndAttributes>()
+                {
+                    {
+                        $"{_tablePrefix}-{WORKFLOW_TABLE}", keys
+                    }
+                }
+            };
+
+            var result = new List<Dictionary<string, AttributeValue>>();
+            BatchGetItemResponse response;
+            do
+            {
+                // Making request
+                response = await _client.BatchGetItemAsync(request);
+
+                // Check the response
+                var responses = response.Responses; // Attribute list in the response.
+                foreach (var tableResponse in responses)
+                {
+                    result.AddRange(tableResponse.Value);
+                }
+
+                // Any unprocessed keys? could happen if you exceed ProvisionedThroughput or some other error.
+                Dictionary<string, KeysAndAttributes> unprocessedKeys = response.UnprocessedKeys;
+                request.RequestItems = unprocessedKeys;
+            } while (response.UnprocessedKeys.Count > 0);
+
+            return result.Select(i => i.ToWorkflowInstance());
+        }
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             subscription.Id = Guid.NewGuid().ToString();
@@ -326,4 +378,4 @@ public void EnsureStoreExists()
             _provisioner.ProvisionTables().Wait();
         }
     }
-}
+}
diff --git a/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs b/src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -75,6 +76,17 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             return JsonConvert.DeserializeObject<WorkflowInstance>(raw, _serializerSettings);
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var raw = await _redis.HashGetAsync($"{_prefix}.{WORKFLOW_SET}", Array.ConvertAll(ids.ToArray(), x => (RedisValue)x));
+            return raw.Select(r => JsonConvert.DeserializeObject<WorkflowInstance>(r, _serializerSettings));
+        }
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             subscription.Id = Guid.NewGuid().ToString();
diff --git a/test/WorkflowCore.UnitTests/BasePersistenceFixture.cs b/test/WorkflowCore.UnitTests/BasePersistenceFixture.cs

Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,7 @@ namespace WorkflowCore.Interface`
`7`	`7`	`{`
`8`	`8`	`/// <remarks>`
`9`	`9`	`/// The implemention of this interface will be responsible for`
`10`		`- /// persisiting running workflow instances to a durable store`
	`10`	`+ /// persisting running workflow instances to a durable store`
`11`	`11`	`/// </remarks>`
`12`	`12`	`public interface IPersistenceProvider`
`13`	`13`	`{`
`@@ -22,6 +22,8 @@ public interface IPersistenceProvider`
`22`	`22`
`23`	`23`	`Task<WorkflowInstance> GetWorkflowInstance(string Id);`
`24`	`24`
	`25`	`+ Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids);`
	`26`	`+`
`25`	`27`	`Task<string> CreateEventSubscription(EventSubscription subscription);`
`26`	`28`
`27`	`29`	`Task<IEnumerable<EventSubscription>> GetSubcriptions(string eventName, string eventKey, DateTime asOf);`
Original file line number	Diff line number	Diff line change
`@@ -8,22 +8,22 @@`
`8`	`8`
`9`	`9`	`namespace WorkflowCore.Services`
`10`	`10`	`{`
`11`		`-`
	`11`	`+`
`12`	`12`	`public interface ISingletonMemoryProvider : IPersistenceProvider`
`13`	`13`	`{`
`14`	`14`	`}`
`15`		`- #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously`
	`15`	`+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously`
`16`	`16`
`17`	`17`	`/// <summary>`
`18`	`18`	`/// In-memory implementation of IPersistenceProvider for demo and testing purposes`
`19`	`19`	`/// </summary>`
`20`	`20`	`public class MemoryPersistenceProvider : ISingletonMemoryProvider`
`21`		`- {`
	`21`	`+ {`
`22`	`22`	`private readonly List<WorkflowInstance> _instances = new List<WorkflowInstance>();`
`23`	`23`	`private readonly List<EventSubscription> _subscriptions = new List<EventSubscription>();`
`24`	`24`	`private readonly List<Event> _events = new List<Event>();`
`25`	`25`	`private readonly List<ExecutionError> _errors = new List<ExecutionError>();`
`26`		`-`
	`26`	`+`
`27`	`27`	`public async Task<string> CreateNewWorkflow(WorkflowInstance workflow)`
`28`	`28`	`{`
`29`	`29`	`lock (_instances)`
`@@ -61,6 +61,19 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)`
`61`	`61`	`}`
`62`	`62`	`}`
`63`	`63`
	`64`	`+ public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)`
	`65`	`+ {`
	`66`	`+ if (ids == null)`
	`67`	`+ {`
	`68`	`+ return new List<WorkflowInstance>();`
	`69`	`+ }`
	`70`	`+`
	`71`	`+ lock (_instances)`
	`72`	`+ {`
	`73`	`+ return _instances.Where(x => ids.Contains(x.Id));`
	`74`	`+ }`
	`75`	`+ }`
	`76`	`+`
`64`	`77`	`public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)`
`65`	`78`	`{`
`66`	`79`	`lock (_instances)`
`@@ -121,7 +134,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)`
`121`	`134`	`}`
`122`	`135`
`123`	`136`	`public void EnsureStoreExists()`
`124`		`- {`
	`137`	`+ {`
`125`	`138`	`}`
`126`	`139`
`127`	`140`	`public async Task<string> CreateEvent(Event newEvent)`
`@@ -133,7 +146,7 @@ public async Task<string> CreateEvent(Event newEvent)`
`133`	`146`	`return newEvent.Id;`
`134`	`147`	`}`
`135`	`148`	`}`
`136`		`-`
	`149`	`+`
`137`	`150`	`public async Task MarkEventProcessed(string id)`
`138`	`151`	`{`
`139`	`152`	`lock (_events)`
`@@ -197,5 +210,5 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)`
`197`	`210`	`}`
`198`	`211`	`}`
`199`	`212`
`200`		`- #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously`
	`213`	`+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously`
`201`	`214`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ public EntityFrameworkPersistenceProvider(IWorkflowDbContextFactory contextFacto`
`26`	`26`	`_canCreateDB = canCreateDB;`
`27`	`27`	`_canMigrateDB = canMigrateDB;`
`28`	`28`	`}`
`29`		`-`
	`29`	`+`
`30`	`30`	`public async Task<string> CreateEventSubscription(EventSubscription subscription)`
`31`	`31`	`{`
`32`	`32`	`using (var db = ConstructDbContext())`
`@@ -96,7 +96,7 @@ public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowSt`
`96`	`96`	`return result;`
`97`	`97`	`}`
`98`	`98`	`}`
`99`		`-`
	`99`	`+`
`100`	`100`	`public async Task<WorkflowInstance> GetWorkflowInstance(string Id)`
`101`	`101`	`{`
`102`	`102`	`using (var db = ConstructDbContext())`
`@@ -115,6 +115,26 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)`
`115`	`115`	`}`
`116`	`116`	`}`
`117`	`117`
	`118`	`+ public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)`
	`119`	`+ {`
	`120`	`+ if (ids == null)`
	`121`	`+ {`
	`122`	`+ return new List<WorkflowInstance>();`
	`123`	`+ }`
	`124`	`+`
	`125`	`+ using (var db = ConstructDbContext())`
	`126`	`+ {`
	`127`	`+ var uids = ids.Select(i => new Guid(i));`
	`128`	`+ var raw = db.Set<PersistedWorkflow>()`
	`129`	`+ .Include(wf => wf.ExecutionPointers)`
	`130`	`+ .ThenInclude(ep => ep.ExtensionAttributes)`
	`131`	`+ .Include(wf => wf.ExecutionPointers)`
	`132`	`+ .Where(x => uids.Contains(x.InstanceId));`
	`133`	`+`
	`134`	`+ return (await raw.ToListAsync()).Select(i => i.ToWorkflowInstance());`
	`135`	`+ }`
	`136`	`+ }`
	`137`	`+`
`118`	`138`	`public async Task PersistWorkflow(WorkflowInstance workflow)`
`119`	`139`	`{`
`120`	`140`	`using (var db = ConstructDbContext())`
`@@ -143,7 +163,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)`
`143`	`163`	`await db.SaveChangesAsync();`
`144`	`164`	`}`
`145`	`165`	`}`
`146`		`-`
	`166`	`+`
`147`	`167`	`public virtual void EnsureStoreExists()`
`148`	`168`	`{`
`149`	`169`	`using (var context = ConstructDbContext())`
`@@ -283,7 +303,7 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)`
`283`	`303`	`}`
`284`	`304`	`}`
`285`	`305`	`}`
`286`		`-`
	`306`	`+`
`287`	`307`	`private WorkflowDbContext ConstructDbContext()`
`288`	`308`	`{`
`289`	`309`	`return _contextFactory.Build();`