State Machine Activities for Kafka Topic Producer and Event Hub Producer

MassTransit · Dec 24, 2020 · f9ad17a · f9ad17a
1 parent 37e4db3
commit f9ad17a
Show file tree

Hide file tree

Showing 24 changed files with 1,134 additions and 97 deletions.
diff --git a/src/Containers/MassTransit.AutofacIntegration/AutofacStateMachineActivityFactory.cs b/src/Containers/MassTransit.AutofacIntegration/AutofacStateMachineActivityFactory.cs
@@ -11,20 +11,12 @@ public class AutofacStateMachineActivityFactory :
     {
         public static readonly IStateMachineActivityFactory Instance = new AutofacStateMachineActivityFactory();
 
-        public Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
             var lifetimeScope = context.GetPayload<ILifetimeScope>();
 
-            return ActivatorUtils.GetOrCreateInstance<TActivity>(lifetimeScope);
-        }
-
-        public Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>
-        {
-            var lifetimeScope = context.GetPayload<ILifetimeScope>();
-
-            return ActivatorUtils.GetOrCreateInstance<TActivity>(lifetimeScope);
+            return ActivatorUtils.GetOrCreateInstance<T>(lifetimeScope);
         }
     }
 }
diff --git a/...xtensionsDependencyInjectionIntegration/DependencyInjectionStateMachineActivityFactory.cs b/...xtensionsDependencyInjectionIntegration/DependencyInjectionStateMachineActivityFactory.cs
@@ -3,7 +3,7 @@
     using System;
     using Automatonymous;
     using GreenPipes;
-    using Metadata;
+    using GreenPipes.Internals.Extensions;
     using Microsoft.Extensions.DependencyInjection;
 
 
@@ -12,25 +12,16 @@ public class DependencyInjectionStateMachineActivityFactory :
     {
         public static readonly IStateMachineActivityFactory Instance = new DependencyInjectionStateMachineActivityFactory();
 
-        Activity<TInstance, TData> IStateMachineActivityFactory.GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-        {
-            return GetActivity<TActivity>(context);
-        }
-
-        Activity<TInstance> IStateMachineActivityFactory.GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-        {
-            return GetActivity<TActivity>(context);
-        }
-
-        static TActivity GetActivity<TActivity>(PipeContext context)
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
             if (context.TryGetPayload(out IServiceScope serviceScope))
-                return serviceScope.ServiceProvider.GetService<TActivity>() ?? ActivatorUtilities.CreateInstance<TActivity>(serviceScope.ServiceProvider);
+                return serviceScope.ServiceProvider.GetService<T>() ?? ActivatorUtilities.CreateInstance<T>(serviceScope.ServiceProvider);
 
             if (context.TryGetPayload(out IServiceProvider serviceProvider))
-                return serviceProvider.GetService<TActivity>() ?? ActivatorUtilities.CreateInstance<TActivity>(serviceProvider);
+                return serviceProvider.GetService<T>() ?? ActivatorUtilities.CreateInstance<T>(serviceProvider);
 
-            throw new PayloadNotFoundException($"IServiceProvider or IServiceScope was not found to create activity: {TypeMetadataCache<TActivity>.ShortName}");
+            throw new PayloadNotFoundException($"IServiceProvider or IServiceScope was not found to get service: {TypeCache<T>.ShortName}");
         }
     }
 }
diff --git a/...ainers/MassTransit.SimpleInjectorIntegration/SimpleInjectorStateMachineActivityFactory.cs b/...ainers/MassTransit.SimpleInjectorIntegration/SimpleInjectorStateMachineActivityFactory.cs
@@ -10,20 +10,12 @@ public class SimpleInjectorStateMachineActivityFactory :
     {
         public static readonly IStateMachineActivityFactory Instance = new SimpleInjectorStateMachineActivityFactory();
 
-        public Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
             var container = context.GetPayload<Scope>();
 
-            return (TActivity)container.GetInstance(typeof(TActivity));
-        }
-
-        public Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>
-        {
-            var container = context.GetPayload<Scope>();
-
-            return (TActivity)container.GetInstance(typeof(TActivity));
+            return container.GetInstance<T>();
         }
     }
 }
diff --git a/...Containers/MassTransit.StructureMapIntegration/StructureMapStateMachineActivityFactory.cs b/...Containers/MassTransit.StructureMapIntegration/StructureMapStateMachineActivityFactory.cs
@@ -10,20 +10,12 @@ public class StructureMapStateMachineActivityFactory :
     {
         public static readonly IStateMachineActivityFactory Instance = new StructureMapStateMachineActivityFactory();
 
-        public Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
-            var lifetimeScope = context.GetPayload<IContainer>();
+            var container = context.GetPayload<IContainer>();
 
-            return lifetimeScope.GetInstance<TActivity>();
-        }
-
-        public Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>
-        {
-            var lifetimeScope = context.GetPayload<IContainer>();
-
-            return lifetimeScope.GetInstance<TActivity>();
+            return container.GetInstance<T>();
         }
     }
 }
diff --git a/src/Containers/MassTransit.WindsorIntegration/WindsorStateMachineActivityFactory.cs b/src/Containers/MassTransit.WindsorIntegration/WindsorStateMachineActivityFactory.cs
@@ -10,20 +10,12 @@ public class WindsorStateMachineActivityFactory :
     {
         public static readonly IStateMachineActivityFactory Instance = new WindsorStateMachineActivityFactory();
 
-        public Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
-            var container = context.GetPayload<IKernel>();
+            var kernel = context.GetPayload<IKernel>();
 
-            return container.Resolve<TActivity>();
-        }
-
-        public Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>
-        {
-            var container = context.GetPayload<IKernel>();
-
-            return container.Resolve<TActivity>();
+            return kernel.Resolve<T>();
         }
     }
 }
diff --git a/src/MassTransit/Automatonymous/Activities/ContainerFactoryActivity.cs b/src/MassTransit/Automatonymous/Activities/ContainerFactoryActivity.cs
@@ -18,7 +18,7 @@ Task Activity<TInstance>.Execute(BehaviorContext<TInstance> context, Behavior<TI
         {
             var factory = context.GetStateMachineActivityFactory();
 
-            Activity<TInstance> activity = factory.GetActivity<TActivity, TInstance>(context);
+            Activity<TInstance> activity = factory.GetService<TActivity>(context);
 
             return activity.Execute(context, next);
         }
@@ -27,7 +27,7 @@ Task Activity<TInstance>.Execute<T>(BehaviorContext<TInstance, T> context, Behav
         {
             var factory = context.GetStateMachineActivityFactory();
 
-            Activity<TInstance> activity = factory.GetActivity<TActivity, TInstance>(context);
+            Activity<TInstance> activity = factory.GetService<TActivity>(context);
 
             var widenBehavior = new WidenBehavior<TInstance, T>(next, context);
 
@@ -69,7 +69,7 @@ void IProbeSite.Probe(ProbeContext context)
         {
             var factory = context.GetStateMachineActivityFactory();
 
-            Activity<TInstance, TData> activity = factory.GetActivity<TActivity, TInstance, TData>(context);
+            Activity<TInstance, TData> activity = factory.GetService<TActivity>(context);
 
             return activity.Execute(context, next);
         }

diff --git a/src/MassTransit/Automatonymous/DefaultConstructorStateMachineActivityFactory.cs b/src/MassTransit/Automatonymous/DefaultConstructorStateMachineActivityFactory.cs
@@ -1,21 +1,16 @@
 namespace Automatonymous
 {
     using System;
+    using GreenPipes;
 
 
     public class DefaultConstructorStateMachineActivityFactory :
         IStateMachineActivityFactory
     {
-        public Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>
+        public T GetService<T>(PipeContext context)
+            where T : class
         {
-            return (Activity<TInstance, TData>)Activator.CreateInstance(typeof(TActivity));
-        }
-
-        public Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>
-        {
-            return (Activity<TInstance>)Activator.CreateInstance(typeof(TActivity));
+            return Activator.CreateInstance<T>();
         }
     }
 }
diff --git a/src/MassTransit/Automatonymous/IStateMachineActivityFactory.cs b/src/MassTransit/Automatonymous/IStateMachineActivityFactory.cs
@@ -1,26 +1,16 @@
 namespace Automatonymous
 {
+    using GreenPipes;
+
+
     public interface IStateMachineActivityFactory
     {
         /// <summary>
-        /// Creates a state machine activity for the specified context
-        /// </summary>
-        /// <typeparam name="TActivity"></typeparam>
-        /// <typeparam name="TInstance"></typeparam>
-        /// <typeparam name="TData"></typeparam>
-        /// <param name="context"></param>
-        /// <returns></returns>
-        Activity<TInstance, TData> GetActivity<TActivity, TInstance, TData>(BehaviorContext<TInstance, TData> context)
-            where TActivity : class, Activity<TInstance, TData>;
-
-        /// <summary>
-        /// Creates a state machine activity for the specified context
+        /// Returns the service, if available, otherwise returns null
         /// </summary>
-        /// <typeparam name="TActivity"></typeparam>
-        /// <typeparam name="TInstance"></typeparam>
-        /// <param name="context"></param>
+        /// <typeparam name="T"></typeparam>
         /// <returns></returns>
-        Activity<TInstance> GetActivity<TActivity, TInstance>(BehaviorContext<TInstance> context)
-            where TActivity : class, Activity<TInstance>;
+        T GetService<T>(PipeContext context)
+            where T : class;
     }
 }
diff --git a/src/MassTransit/Automatonymous/PublishExtensions.cs b/src/MassTransit/Automatonymous/PublishExtensions.cs
@@ -84,7 +84,7 @@ public static class PublishExtensions
             where TMessage : class
             where TException : Exception
         {
-            return source.Add(new PublishActivity<TInstance, TMessage>(x => message, contextCallback));
+            return source.Add(new FaultedPublishActivity<TInstance, TException, TMessage>(x => message, contextCallback));
         }
 
         public static ExceptionActivityBinder<TInstance, TException> PublishAsync<TInstance, TException, TMessage>(
@@ -94,7 +94,7 @@ public static class PublishExtensions
             where TMessage : class
             where TException : Exception
         {
-            return source.Add(new PublishActivity<TInstance, TMessage>(x => message, contextCallback));
+            return source.Add(new FaultedPublishActivity<TInstance, TException, TMessage>(x => message, contextCallback));
         }
 
         public static ExceptionActivityBinder<TInstance, TException> Publish<TInstance, TException, TMessage>(
@@ -127,7 +127,7 @@ public static class PublishExtensions
             where TMessage : class
             where TException : Exception
         {
-            return source.Add(new PublishActivity<TInstance, TData, TMessage>(x => message, contextCallback));
+            return source.Add(new FaultedPublishActivity<TInstance, TData, TException, TMessage>(x => message, contextCallback));
         }
 
         public static ExceptionActivityBinder<TInstance, TData, TException> PublishAsync<TInstance, TData, TException, TMessage>(
@@ -138,7 +138,7 @@ public static class PublishExtensions
             where TMessage : class
             where TException : Exception
         {
-            return source.Add(new PublishActivity<TInstance, TData, TMessage>(x => message, contextCallback));
+            return source.Add(new FaultedPublishActivity<TInstance, TData, TException, TMessage>(x => message, contextCallback));
         }
 
         public static ExceptionActivityBinder<TInstance, TData, TException> Publish<TInstance, TData, TException, TMessage>(

diff --git a/src/MassTransit/Exceptions/ProduceException.cs b/src/MassTransit/Exceptions/ProduceException.cs
@@ -0,0 +1,30 @@
+namespace MassTransit
+{
+    using System;
+    using System.Runtime.Serialization;
+
+
+    [Serializable]
+    public class ProduceException :
+        MassTransitException
+    {
+        public ProduceException()
+        {
+        }
+
+        public ProduceException(string message)
+            : base(message)
+        {
+        }
+
+        public ProduceException(string message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
+
+        protected ProduceException(SerializationInfo info, StreamingContext context)
+            : base(info, context)
+        {
+        }
+    }
+}