V1/feature/121

src/Cortex.Mediator/DependencyInjection/MediatorOptions.cs

@@ -1,4 +1,4 @@
-using Cortex.Mediator.Commands;
+using Cortex.Mediator.Commands;
 using Cortex.Mediator.Queries;
 using System;
 using System.Collections.Generic;
@@ -11,6 +11,8 @@ public class MediatorOptions
         internal List<Type> CommandBehaviors { get; } = new();
         internal List<Type> QueryBehaviors { get; } = new();
 
+        public bool OnlyPublicClasses { get; set; } = true;
+
         public MediatorOptions AddCommandPipelineBehavior<TBehavior>()
             where TBehavior : ICommandPipelineBehavior<ICommand> // Add constraint
         {

src/Cortex.Mediator/DependencyInjection/ServiceCollectionExtensions.cs

@@ -1,4 +1,4 @@
-using Cortex.Mediator.Commands;
+using Cortex.Mediator.Commands;
 using Cortex.Mediator.Infrastructure;
 using Cortex.Mediator.Notifications;
 using Cortex.Mediator.Queries;
@@ -27,36 +27,37 @@ public static IServiceCollection AddCortexMediator(
             services.AddValidatorsFromAssemblies(handlerAssemblyMarkerTypes.Select(t => t.Assembly));
             services.AddUnitOfWork();
 
-            RegisterHandlers(services, handlerAssemblyMarkerTypes);
+            RegisterHandlers(services, handlerAssemblyMarkerTypes, options);
             RegisterPipelineBehaviors(services, options);
 
             return services;
         }
 
         private static void RegisterHandlers(
             IServiceCollection services,
-            IEnumerable<Type> assemblyMarkerTypes)
+            IEnumerable<Type> assemblyMarkerTypes,
+            MediatorOptions options)
         {
             var assemblies = assemblyMarkerTypes.Select(t => t.Assembly).ToArray();
 
             services.Scan(scan => scan
                 .FromAssemblies(assemblies)
                 .AddClasses(classes => classes
-                    .AssignableTo(typeof(ICommandHandler<>)))
+                    .AssignableTo(typeof(ICommandHandler<>)), options.OnlyPublicClasses)
                 .AsImplementedInterfaces()
                 .WithScopedLifetime());
 
             services.Scan(scan => scan
                 .FromAssemblies(assemblies)
                 .AddClasses(classes => classes
-                    .AssignableTo(typeof(IQueryHandler<,>)))
+                    .AssignableTo(typeof(IQueryHandler<,>)), options.OnlyPublicClasses)
                 .AsImplementedInterfaces()
                 .WithScopedLifetime());
 
             services.Scan(scan => scan
                 .FromAssemblies(assemblies)
                 .AddClasses(classes => classes
-                    .AssignableTo(typeof(INotificationHandler<>)))
+                    .AssignableTo(typeof(INotificationHandler<>)), options.OnlyPublicClasses)
                 .AsImplementedInterfaces()
                 .WithScopedLifetime());
         }
@@ -76,7 +77,6 @@ private static void RegisterPipelineBehaviors(IServiceCollection services, Media
             }
         }
 
-
         private static void AddUnitOfWork(this IServiceCollection services)
         {
             services.AddScoped<IUnitOfWork>(provider =>

src/Cortex.Streams.Pulsar/PulsarSinkOperator.cs

@@ -28,6 +28,10 @@ public PulsarSinkOperator(string serviceUrl, string topic, ISerializer<TInput> s
             _client = PulsarClient.Builder()
                 .ServiceUrl(new Uri(_serviceUrl))
                 .Build();
+
+            // BUG #103 Start PulsarSink Operator when Sink is initialized
+            // Pulsar Producer doesnot start when the production happens, we have to start the Producer when it is initialized.
+            Start();
         }
 
         public void Start()