3
\$\begingroup\$

I have implemented a event aggregator for our backend.

public class EventAggregator : IEventAggregator
{
    private readonly WeakReferenceList<object> _subscribers = new WeakReferenceList<object>();

    public void Subscribe(object subscriber)
    {
        _subscribers.Add(subscriber);
    }

    public Task PublishAsync<T>(T message) where T : class
    {
        foreach (var handler in _subscribers.OfType<IHandle<T>>())
            handler.Handle(message);

        var handlers = _subscribers
            .OfType<IHandleAsync<T>>()
            .Select(s => s.HandleAsync(message))
            .Where(t => t.Status != TaskStatus.RanToCompletion)
            .ToList();

        if (handlers.Any()) return Task.WhenAll(handlers);
        return Task.CompletedTask;
    }
}

You can then choose to handle the events async or none async

None async

public class CreatedListener : IBusinessContextListener, IHandle<SavingChangesEvent>
{
    public CreatedListener(IEventAggregator eventAggregator)
    {
        eventAggregator.Subscribe(this);
    }

    public void Handle(SavingChangesEvent message)
    {
        foreach (var created in message.Context.Db.ChangeTracker.Entries<ICreated>().Where(c => c.State == EntityState.Added))
        {
            created.Entity.CreatedBy = message.Context.Username;
            created.Entity.CreatedUTC = DateTime.UtcNow;
        }
    }
}

Async

public class AzureFileStorageExternalAdapter : IExternalPartAdapter, IHandleAsync<TransactionCommitedEvent>
{
    private readonly IBusinessContext _ctx;
    private readonly List<CloudFile> _proccessedFiles;

    public AzureFileStorageExternalAdapter(IBusinessContext ctx, IEventAggregator eventAggregator)
    {
        _ctx = ctx;
        _proccessedFiles = new List<CloudFile>();
        eventAggregator.Subscribe(this);
    }


    public Task<IEnumerable<byte[]>> ListResponses()
    {
        var folder = GetRemoteFolder($"In:{_ctx.ExecutionContext.Name}");
        var files = folder.ListFilesAndDirectories()
            .Select(fi => fi as CloudFile)
            .Where(file => file != null)
            .ToList();

        _proccessedFiles.AddRange(files);

        return files
            .Select(async file =>
            {
                using (var stream = await file.OpenReadAsync())
                {
                    using (var mem = new MemoryStream())
                    {
                        await stream.CopyToAsync(mem);
                        mem.Position = 0;
                        return mem.ToArray();
                    }
                }
            })
            .WhenAll();
    }


    public Task HandleAsync(TransactionCommitedEvent message)
    {
        if (!_proccessedFiles.Any()) return Task.CompletedTask;

        return _proccessedFiles
            .Select(file => file.DeleteAsync())
            .WhenAll();
    }
}

Publisher has to be async even though only none async listeners are currently listenting

public class BusinessContext : IBusinessContextController, IDisposable
{
    private readonly IEventAggregator _eventAggregator;
    private IDbContextTransaction _dbContextTransaction;

    public BusinessContext(DbContext ctx, IEnumerable<IBusinessContextListener> ctxListeners, IEventAggregator eventAggregator)
    {
        _eventAggregator = eventAggregator;
        Db = ctx;
        if(ctxListeners == null) throw new ApplicationException("Do not remove this dependency, it ensures that listeners are awake");
    }

    public DbContext Db { get; }

    public string Username { get; set; }

    public ExecutionContext ExecutionContext { get; set; }

    public async Task SaveChangesAsync()
    {
        await _eventAggregator.PublishAsync(new SavingChangesEvent(this));
        await Db.SaveChangesAsync();
    }

    public async Task StartTransactionAsync()
    {
        if (_dbContextTransaction != null) return;

        _dbContextTransaction = await Db.Database.BeginTransactionAsync();
    }

    public void Dispose()
    {
        _dbContextTransaction?.Dispose();
    }

    public Task CommitTransactionAsync()
    {
        _dbContextTransaction?.Commit();
        return _eventAggregator.PublishAsync(new TransactionCommitedEvent());
    }

    public void RollbackTransaction()
    {
        _dbContextTransaction?.Rollback();
    }
}

Bonus question the interface IBusinessContextListener is a empty markup interface that is only used to mark classes that have no other purpose than to be a event listener (Otherwise they would not be 'awake' and there Handle methods would not invoke. Solid design?

\$\endgroup\$
2
\$\begingroup\$

Scheduling & Latency

One single handler could block all other handlers by performing a long-running calculation:

public void Handle(SavingChangesEvent message)
{
    Thread.Sleep(10000); // long running code ..
}

While this is the exact same behavior of the default Event Pattern in C#, you might expect an aggregator to be able to work around this - especially since the method is called PublishAsync. It's not the best design decision to have this method perform both synchronous IHandle<T> and asynchronous IHandleAsync<T> event handler notifications.

In addition, if you have a single listener on EventT1 and 1000 listeners on EventT2, you'd always loop all those listeners (even twice in your code) to find the listeners OfType<RequestedType>(). Using one container list for all listeners is something I would try to avoid. A dictionary by Type and its listeners is a better approach.

Thread-Safety

While WeakReferenceList is thread-safe when adding and removing items, enumerating the items while adding, removing items concurrently is not. You'd have to implement a custom threading mechanism that safeguards against registration during event notification.

Error-Handling

Any error in a synchronous-aware handler exits notifications early. Since the synchronous notifications take priority over the asynchronous ones, these won't even be notified. Perhaps implementing an UnobservedExceptionHandler could help you make a robust design.

Micro-optimisations

Filtering out Where(t => t.Status != TaskStatus.RanToCompletion) and returning Task.CompletedTask when handler.Any() is false, is not going to gain you much. In fact, in between checking the status and returning the completed task, the status may have already been changed. I would disregard the status and always return return Task.WhenAll(handlers). Let the waiters handle the status, which is built in using async/await.

Ambiguously-defined handlers

Handlers that implement both interfaces are not able to define which of the interfaces they want to see handled. By default, both their handlers will be notified. Is this as designed or should more fine-grained registration be allowed?

Weak Reference Pattern

I am not sure using weak references is the best design decision here. According to Microsoft:

Avoid using weak references as an automatic solution to memory management problems. Instead, develop an effective caching policy for handling your application's objects.

And I would agree. Instead I would use registration and deregistration methods.

\$\endgroup\$
  • 2
    \$\begingroup\$ I should remove that weak reference list, we use this aggregator on a per request lifetime using DI so it never lives longer than a request. Maintainability over performance, if we run into performance we can easily refactor for performance. \$\endgroup\$ – Anders Aug 12 at 17:49
  • \$\begingroup\$ I would say that is a good call :) \$\endgroup\$ – dfhwze Aug 12 at 17:49
  • 1
    \$\begingroup\$ Got and upvote so revisited and I noticed you favor Type lookup, our IHandle interface is contravariant so you can handle a set of messages if you listen to the baseclass since OfType works with contravariances. You need to implement this with your type look up. Either by lazy lookup with IsAssignableFrom and append to type lookup, or by iterating all types in assembly at startup and add them to the lookup \$\endgroup\$ – Anders Sep 4 at 10:14

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.