20181227.simple_task_actor_full.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Castle.DynamicProxy;
using Fasterflect;
using Nito.AsyncEx;

namespace ConsoleApplication10
{
    public class SecurityEvent
    {
        public SecurityEvent(string content)
        {
            this.Content = content;
        }

        public string Content { get; }
    }

    public interface IActor
    {
        void NotifyTo<TEvent>(IActor actor);
    }

    public interface IPerson : IActor
    {
        Task<string> GetNameAsync(int id);
        Task<int> GetAgeAsync(int id);
    }

    public interface ISomeSubscriber : IActor
    {
        Task Event(SecurityEvent @event);
    }

    public abstract class Actor : IActor
    {
        private class Subscriber
        {
            private readonly object instance;
            private readonly MethodInvoker methodInvoker;

            internal Subscriber(object instance, Type type)
            {
                this.instance = instance;
                this.methodInvoker = instance.GetType().DelegateForCallMethod("Event", type);
            }

            internal void Invoke(object @event)
            {
                this.methodInvoker.Invoke(this.instance, @event);
            }
        }

        private readonly Dictionary<Type, IList<Subscriber>> subscriptions = new Dictionary<Type, IList<Subscriber>>();

        public void NotifyTo<TEvent>(IActor actor)
        {
            IList<Subscriber> mailboxList;

            var subscriber = new Subscriber(actor, typeof(TEvent));

            if (this.subscriptions.TryGetValue(typeof(Subscriber), out mailboxList))
            {
                mailboxList.Add(subscriber);
            }
            else
            {
                this.subscriptions.Add(typeof(TEvent), new List<Subscriber> { subscriber });
            }
        }

        public void Publish(object @event)
        {
            var subscriberOfInterest = this.subscriptions.FirstOrDefault(i => i.Key.IsInstanceOfType(@event)).Value;

            foreach (var subscriber in subscriberOfInterest)
            {
                Task.Run(() => subscriber.Invoke(@event));
            }
        }
    }

    public class Person : Actor, IPerson
    {
        // Generated with https://www.fakenamegenerator.com/gen-random-gr-gr.php
        private readonly string[] names = {
        "Brigitte Freytag",
        "Marina Hoffmann",
        "Lea Wagner",
        "Frank Diederich",
        "Nadine Fuchs"
    };

        public async Task<string> GetNameAsync(int id)
        {
            ColoredConsole.WriteLine($"{nameof(this.GetNameAsync)} with id {id} ...");

            // Takes a long time ...
            await Task.Delay(3000);

            this.Publish(new SecurityEvent($"Id {id} was successfuly requested!"));

            return this.names[id];
        }

        public async Task<int> GetAgeAsync(int id)
        {
            ColoredConsole.WriteLine($"{nameof(this.GetAgeAsync)} with id {id} ...");

            await Task.Delay(2000);

            return id * 10;
        }
    }

    public class SomeSubscriber : Actor, ISomeSubscriber
    {
        public Task Event(SecurityEvent @event)
        {
            ColoredConsole.WriteLine(@event.Content + " Start encrypting ...", ConsoleColor.Red);

            // Processing security event is complicated ...
            Thread.Sleep(6000);

            ColoredConsole.WriteLine(@event.Content + " Finished!", ConsoleColor.Red);

            return Task.CompletedTask;
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var personActor = CreateActor<IPerson, Person>(new Person());
            var subscriberActor = CreateActor<ISomeSubscriber, SomeSubscriber>(new SomeSubscriber());

            personActor.NotifyTo<SecurityEvent>(subscriberActor);

            for (int i = 0; i < 5; i++)
            {
                var localI = i;

                new Thread(() =>
                {
                    var nameTask = personActor.GetNameAsync(localI);
                    var ageTask = personActor.GetAgeAsync(localI);

                    Task.WaitAll(nameTask, ageTask);

                    ColoredConsole.WriteLine($"{DateTime.Now} {nameTask.Result}, {ageTask.Result}", ConsoleColor.Green);

                }).Start();
            }

            Console.ReadLine();
        }

        private static TActorType CreateActor<TActorType, TConcreteType>(TActorType instance) where TActorType : class
        {
            var proxyGenerator = new ProxyGenerator();

            return proxyGenerator.CreateInterfaceProxyWithTarget(
                instance,
                ProxyGenerationOptions.Default,
                new ActorInterceptor<TConcreteType>());
        }
    }

    internal class ActorInterceptor<TActor> : IAsyncInterceptor
    {
        private readonly AsyncLock mutex = new AsyncLock();
        private readonly Dictionary<MethodInfo, MethodInvoker> cache = new Dictionary<MethodInfo, MethodInvoker>();

        public ActorInterceptor()
        {
            foreach (var @interface in typeof(TActor).GetInterfaces())
            {
                foreach (var method in @interface.GetMethods())
                {
                    var methodInvoker = typeof(TActor).DelegateForCallMethod(method.Name, method.Parameters().Select(i => i.ParameterType).ToArray());

                    this.cache.Add(method, methodInvoker);
                }
            }
        }

        public void InterceptSynchronous(IInvocation invocation)
        {
            invocation.Proceed();
        }

        public void InterceptAsynchronous(IInvocation invocation)
        {
            invocation.ReturnValue = this.InternalInterceptAsynchronous(invocation);
        }

        public void InterceptAsynchronous<TResult>(IInvocation invocation)
        {
            invocation.ReturnValue = this.InternalInterceptAsynchronous<TResult>(invocation);
        }

        private async Task<TResult> InternalInterceptAsynchronous<TResult>(IInvocation invocation)
        {
            using (await this.mutex.LockAsync())
            {
                MethodInvoker methodInvoker;

                if (this.cache.TryGetValue(invocation.Method, out methodInvoker))
                {
                    return await ((Task<TResult>)methodInvoker.Invoke(invocation.InvocationTarget, invocation.Arguments)).ConfigureAwait(false);
                }

                throw new MissingMethodException("Method missing");
            }
        }

        private async Task InternalInterceptAsynchronous(IInvocation invocation)
        {
            using (await this.mutex.LockAsync())
            {
                MethodInvoker methodInvoker;

                if (this.cache.TryGetValue(invocation.Method, out methodInvoker))
                {
                    await ((Task)methodInvoker.Invoke(invocation.InvocationTarget, invocation.Arguments)).ConfigureAwait(false);
                }
            }
        }
    }

    public static class ColoredConsole
    {
        private static readonly object SyncRoot = new object();

        public static void WriteLine(string content, ConsoleColor color = ConsoleColor.Gray)
        {
            lock (SyncRoot)
            {
                Console.ForegroundColor = color;
                Console.WriteLine($"{DateTime.Now} [{Thread.CurrentThread.ManagedThreadId}] {content}");
                Console.ResetColor();
            }
        }
    }
}