Add some concepts

Author: fembina

Examples/Falko.Talkie.Examples.Microsoft.Hosting/ExBehaviour.cs

@@ -0,0 +1,32 @@
+using Talkie.Common;
+using Talkie.Controllers.MessageControllers;
+using Talkie.Disposables;
+using Talkie.Flows;
+using Talkie.Handlers;
+using Talkie.Models.Messages;
+using Talkie.Pipelines.Handling;
+using Talkie.Pipelines.Intercepting;
+using Talkie.Signals;
+using Talkie.Subscribers;
+
+namespace Talkie.Examples;
+
+public sealed class ExBehaviour : IBehaviorsSubscriber
+{
+    public void Subscribe(ISignalFlow flow, IRegisterOnlyDisposableScope disposables, CancellationToken cancellationToken)
+    {
+        var s = ConcurrentExclusiveSchedulerPair
+
+        flow.Subscribe(signals => signals
+            .OfType<UserAcceptInvoice>()
+            .SkipSelfPublished()
+            .Where(signal => signal.Message.GetText().IsNullOrEmpty() is false)
+            .HandleAsync(async context =>
+            {
+                Console.WriteLine("Incoming message with text");
+
+                await context.ToMessageController().PublishMessageAsync("Hello");
+            }))
+            .UnsubscribeWith(disposables);
+    }
+}

Examples/Falko.Talkie.Examples.Microsoft.Hosting/HelloSubscriber.cs

@@ -17,8 +17,10 @@ public void Subscribe(ISignalFlow flow, IRegisterOnlyDisposableScope disposables
     {
         flow.Subscribe<MessagePublishedSignal>(static signals => signals
             .SkipSelfPublished()
-            .SkipOlderThan(TimeSpan
-                .FromMinutes(1))
+            .SkipOlderThan(TimeSpan.FromMinutes(1))
+            .InterceptOn(Schedulers.Intercepting.Random())
+            .Do(_ => Thread.Sleep(1000))
+            .InterceptOn(Schedulers.Intercepting.Current())
             .Where(signal => signal
                 .Message
                 .GetText()

Examples/Falko.Talkie.Examples.Microsoft.Hosting/Program.cs

@@ -1,21 +1,15 @@
-using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.Hosting;
-using Serilog;
-using Talkie.Examples;
-using Talkie.Hosting;
+// IWith и в чем отличие от обычных интерфейсов
+// Sequence/FrozenSequence и отличие от List
+// IDisposable и IAsyncDisposable, и IDisposableScope и IRegisterOnlyDisposableScope
+// foreach
+// Parallelize().ForEach
+// async/await
 
-// Please set 'Telegram:Token' environment variable before running 💚💚💚
+using Talkie.Bridges.Telegram.Clients;
+using Talkie.Bridges.Telegram.Configurations;
 
-await new HostBuilder()
-    .ConfigureAppConfiguration(configuration => configuration
-        .AddEnvironmentVariables())
-    .UseSerilog((_, configuration) => configuration
-        .MinimumLevel.Verbose()
-        .WriteTo.Console())
-    .UseTalkie(configuration => configuration
-        .SetSignalsLogging())
-    .ConfigureServices(services => services
-        .AddIntegrations<TelegramSubscriber>()
-        .AddBehaviors<HelloSubscriber>()
-        .AddBehaviors<StartSubscriber>())
-    .RunConsoleAsync();
+var configuration = new TelegramConfiguration("TOKEN");
+
+var client = new TelegramClient(configuration);
+
+var self = await client.GetMeAsync();

Sources/Falko.Talkie.Signals.Extensions/Pipelines/Handling/SignalHandlingPipelineBuilderExtensions.On.cs

@@ -0,0 +1,44 @@
+using Talkie.Pipelines.Intercepting;
+using Talkie.Signals;
+
+namespace Talkie.Pipelines.Handling;
+
+public static partial class SignalHandlingPipelineBuilderExtensions
+{
+    public static ISignalHandlingPipelineBuilder HandleOn
+    (
+        this ISignalInterceptingPipelineBuilder builder,
+        ISignalHandlingPipelineProcessorFactory processorFactory
+    )
+    {
+        var pipeline = builder.Build();
+
+        return pipeline is not EmptySignalInterceptingPipeline
+            ? new SignalHandlingPipelineBuilder(pipeline)
+            : SignalHandlingPipelineBuilder.Empty;
+    }
+
+    public static ISignalHandlingPipelineBuilder<T> HandleOn<T>
+    (
+        this ISignalInterceptingPipelineBuilder<T> builder,
+        ISignalHandlingPipelineProcessorFactory processorFactory
+    ) where T : Signal
+    {
+        var pipeline = builder.Build();
+
+        return pipeline is not EmptySignalInterceptingPipeline
+            ? new SignalHandlingPipelineBuilder<T>(pipeline)
+            : SignalHandlingPipelineBuilder<T>.Empty;
+    }
+
+    public static ISignalInterceptingPipelineBuilder<T> InterceptOn<T>
+    (
+        this ISignalInterceptingPipelineBuilder<T> builder,
+        ISignalInterceptingPipelineProcessorFactory processorFactory
+    ) where T : Signal
+    {
+        var pipeline = builder.Build();
+
+        return builder;
+    }
+}

Sources/Falko.Talkie.Signals/Concurrent/Schedulers.cs

@@ -0,0 +1,46 @@
+using Talkie.Pipelines.Handling;
+using Talkie.Pipelines.Intercepting;
+
+namespace Talkie.Concurrent;
+
+public static partial class Schedulers
+{
+    public static readonly HandlingSchedulers Handling = default;
+
+    public static readonly InterceptingSchedulers Intercepting = default;
+}
+
+public readonly struct HandlingSchedulers;
+
+public readonly struct InterceptingSchedulers;
+
+public static partial class HandlingSchedulersExtensions
+{
+    public static ISignalHandlingPipelineProcessorFactory Current(this HandlingSchedulers _)
+    {
+        throw new NotImplementedException();
+    }
+
+    public static ISignalHandlingPipelineProcessorFactory Random(this HandlingSchedulers _)
+    {
+        throw new NotImplementedException();
+    }
+
+    public static ISignalHandlingPipelineProcessorFactory Parallel(this HandlingSchedulers _)
+    {
+        return ParallelSignalHandlingPipelineSchedulerFactory.Instance;
+    }
+}
+
+public static partial class InterceptingSchedulersExtensions
+{
+    public static ISignalInterceptingPipelineProcessorFactory Current(this InterceptingSchedulers _)
+    {
+        throw new NotImplementedException();
+    }
+
+    public static ISignalInterceptingPipelineProcessorFactory Random(this InterceptingSchedulers _)
+    {
+        throw new NotImplementedException();
+    }
+}

Sources/Falko.Talkie.Signals/Pipelines/Handling/ISignalHandlingPipelineProcessor.cs

@@ -0,0 +1,12 @@
+using Talkie.Handlers;
+
+namespace Talkie.Pipelines.Handling;
+
+public interface ISignalHandlingPipelineProcessor
+{
+    ValueTask ProcessAsync
+    (
+        SignalContext context,
+        CancellationToken cancellationToken
+    );
+}

Sources/Falko.Talkie.Signals/Pipelines/Handling/ISignalHandlingPipelineProcessorFactory.cs

@@ -0,0 +1,9 @@
+using Talkie.Handlers;
+using Talkie.Sequences;
+
+namespace Talkie.Pipelines.Handling;
+
+public interface ISignalHandlingPipelineProcessorFactory
+{
+    ISignalHandlingPipelineProcessor Create(FrozenSequence<ISignalHandler> handlers);
+}

Sources/Falko.Talkie.Signals/Pipelines/Handling/ParallelSignalHandlingPipelineProcessor.cs

@@ -0,0 +1,18 @@
+using Talkie.Concurrent;
+using Talkie.Handlers;
+using Talkie.Sequences;
+
+namespace Talkie.Pipelines.Handling;
+
+public sealed class ParallelSignalHandlingPipelineProcessor(FrozenSequence<ISignalHandler> handlers) : ISignalHandlingPipelineProcessor
+{
+    private readonly ParallelismMeter _handlersParallelismMeter = new();
+
+    public ValueTask ProcessAsync(SignalContext context, CancellationToken cancellationToken)
+    {
+        return handlers.Parallelize(_handlersParallelismMeter)
+            .ForEachAsync((handler, scopedCancellationToken) => handler.HandleAsync(context, scopedCancellationToken),
+                cancellationToken: cancellationToken)
+            .AsValueTask();
+    }
+}

Sources/Falko.Talkie.Signals/Pipelines/Handling/ParallelSignalHandlingPipelineSchedulerFactory.cs

@@ -0,0 +1,16 @@
+using Talkie.Handlers;
+using Talkie.Sequences;
+
+namespace Talkie.Pipelines.Handling;
+
+public sealed class ParallelSignalHandlingPipelineSchedulerFactory : ISignalHandlingPipelineProcessorFactory
+{
+    public static readonly ParallelSignalHandlingPipelineSchedulerFactory Instance = new();
+
+    private ParallelSignalHandlingPipelineSchedulerFactory() { }
+
+    public ISignalHandlingPipelineProcessor Create(FrozenSequence<ISignalHandler> handlers)
+    {
+        return new ParallelSignalHandlingPipelineProcessor(handlers);
+    }
+}

Sources/Falko.Talkie.Signals/Pipelines/Handling/SequencingHandlingPipelineProcessor.cs

@@ -0,0 +1,15 @@
+using Talkie.Handlers;
+using Talkie.Sequences;
+
+namespace Talkie.Pipelines.Handling;
+
+public sealed class SequencingHandlingPipelineProcessor(FrozenSequence<ISignalHandler> handlers) : ISignalHandlingPipelineProcessor
+{
+    public async ValueTask ProcessAsync(SignalContext context, CancellationToken cancellationToken)
+    {
+        foreach (var handler in handlers.AsEnumerable())
+        {
+            await handler.HandleAsync(context, cancellationToken);
+        }
+    }
+}