Lokad.Cqrs.SimpleSynchronousMessageSender

ExtendCqrsClientBuilder.cs

using System;

using Lokad.Cqrs.Build.Client;

namespace Lokad.Cqrs.Extensions.Build
{
    public static class ExtendCqrsClientBuilder
    {
        public static CqrsClientBuilder AddSynchronousMessageSender(this CqrsClientBuilder builder)
        {
            return AddSynchronousMessageSender(builder, config => { });
        }

        public static CqrsClientBuilder AddSynchronousMessageSender(this CqrsClientBuilder builder,
                                                                    Action<SynchronousSendMessageModule> config)
        {
            var module = new SynchronousSendMessageModule();
            config(module);
            builder.EnlistModule(module);
            return builder;
        }
    }
}

ISynchronousMessageSender.cs

using System;

using Lokad.Cqrs.Core.Envelope;

namespace Lokad.Cqrs.Extensions
{
    public interface ISynchronousMessageSender
    {
        TEntity Send<TKey, TEntity>(TKey entityKey, object content);
        TEntity Send<TKey, TEntity>(TKey entityKey, object content, Action<EnvelopeBuilder> configure);
    }
}

SimpleSynchronousMessageSender.cs

using System;
using System.Threading;

using Autofac;

using Lokad.Cqrs.Core.Envelope;
using Lokad.Cqrs.Feature.AtomicStorage;

namespace Lokad.Cqrs.Extensions
{
    internal class SimpleSynchronousMessageSender : ISynchronousMessageSender
    {
        private readonly TimeSpan checkInterval;
        private readonly ILifetimeScope scope;
        private readonly IMessageSender sender;

        public SimpleSynchronousMessageSender(IMessageSender sender, ILifetimeScope scope, TimeSpan checkInterval)
        {
            this.sender = sender;
            this.scope = scope;
            this.checkInterval = checkInterval;
        }

        #region Implementation of ISynchronousMessageSender

        public TEntity Send<TKey, TEntity>(TKey entityKey, object content)
        {
            return Send<TKey, TEntity>(entityKey, content, builder => { });
        }

        public TEntity Send<TKey, TEntity>(TKey entityKey, object content, Action<EnvelopeBuilder> configure)
        {
            sender.SendOne(content, configure);

            return WaitForEntity<TKey, TEntity>(entityKey);
        }

        private T1 WaitForEntity<T, T1>(T entityKey)
        {
            var reader = scope.Resolve<IAtomicEntityReader<T, T1>>();

            T1 entity;
            while (!reader.TryGet(entityKey, out entity))
            {
                Thread.Sleep(checkInterval);
            }

            return entity;
        }

        #endregion
    }
}

SynchronousSendMessageModule.cs

using System;

using Autofac;
using Autofac.Core;

using Lokad.Cqrs.Core;

namespace Lokad.Cqrs.Extensions.Build
{
    public sealed class SynchronousSendMessageModule : HideObjectMembersFromIntelliSense, IModule
    {
        private TimeSpan checkInterval = TimeSpan.FromSeconds(1);

        #region IModule Members

        public void Configure(IComponentRegistry componentRegistry)
        {
            componentRegistry.Register(BuildMessageSender);
        }

        #endregion

        public void CheckEvery(TimeSpan interval)
        {
            checkInterval = interval;
        }

        private ISynchronousMessageSender BuildMessageSender(IComponentContext c)
        {
            var sender = c.Resolve<IMessageSender>();
            var scope = c.Resolve<ILifetimeScope>();
            return new SimpleSynchronousMessageSender(sender, scope, checkInterval);
        }
    }
}