Manual NServiceBus Configuration

In my previous post I have shown a simple implementation of Publish / Subscribe messages using the NServiceBus. We may notice from the example provided that to run the Service Bus we executed configuration through a series of extesion methods supplied.

Bus = Configure.With().DefaultBuilder().BinarySerializer()
         .MsmqTransport().IsTransactional(true).PurgeOnStartup(false)
         .UnicastBus().ImpersonateSender(false).LoadMessageHandlers()
         .CreateBus()
         .Start();

I managed to explore more and look into how each extension methods are performing the configuration. Here is what I discovered.

Configure config = Configure.With();

// Setting the DefaultBuilder
IContainer container = new SpringObjectBuilder();
CommonObjectBuilder b = new CommonObjectBuilder { Container = container, Synchronized = SyncConfig.Synchronize };
config.Builder = b;
config.Configurer = b;
config.Configurer.ConfigureComponent<CommonObjectBuilder>(ComponentCallModelEnum.Singleton)
    .ConfigureProperty(c => c.Container, container);
SyncConfig.MarkConfigured();

// Setting the Serialization
config.Configurer.ConfigureComponent<SimpleMessageMapper>(ComponentCallModelEnum.Singleton);
config.Configurer.ConfigureComponent(typeof(MessageSerializer), ComponentCallModelEnum.Singleton);

// Setting MSMQ Transport
ConfigMsmqTransport transportCfg = new ConfigMsmqTransport();
transportCfg.Configure(config);
transportCfg.IsTransactional(true);
transportCfg.PurgeOnStartup(false);

// Setting the Unicast Bus
ConfigUnicastBus unicastCfg = new ConfigUnicastBus();
unicastCfg.Configure(config);
unicastCfg.ImpersonateSender(true);
unicastCfg.LoadMessageHandlers();

// Creating the Bus
IStartableBus startableBus = config.CreateBus();

// Staring the Bus
Bus = startableBus.Start();

Digging further into the details we can slice the MSMQ configuration using the following code.

IComponentConfig<MsmqTransport> transportCfg = config.Configurer.ConfigureComponent<MsmqTransport>(ComponentCallModelEnum.Singleton);
MsmqTransportConfig msmqTransportCfg = Configure.GetConfigSection<MsmqTransportConfig>();
if (msmqTransportCfg != null)
{
    transportCfg.ConfigureProperty(t => t.InputQueue, msmqTransportCfg.InputQueue);
    transportCfg.ConfigureProperty(t => t.NumberOfWorkerThreads, msmqTransportCfg.NumberOfWorkerThreads);
    transportCfg.ConfigureProperty(t => t.ErrorQueue, msmqTransportCfg.ErrorQueue);
    transportCfg.ConfigureProperty(t => t.MaxRetries, msmqTransportCfg.MaxRetries);
    transportCfg.ConfigureProperty(t => t.IsTransactional, true);
    transportCfg.ConfigureProperty(t => t.PurgeOnStartup, false);
}

The Unicast settings can be further broken into the following code

IComponentConfig<UnicastBus> unicastCfg = config.Configurer.ConfigureComponent<UnicastBus>(ComponentCallModelEnum.Singleton);
Type authType = Configure.TypesToScan.Where(t => typeof(IAuthorizeSubscriptions).IsAssignableFrom(t) && !t.IsInterface).FirstOrDefault();
if (authType != null)
    config.Configurer.ConfigureComponent(authType, ComponentCallModelEnum.Singleton);
Configure.TypesToScan.Where(t => typeof(IMessageModule).IsAssignableFrom(t) && !t.IsInterface).ToList().ForEach(
        type => config.Configurer.ConfigureComponent(type, ComponentCallModelEnum.Singleton)
    );
UnicastBusConfig unicastBusCfg = Configure.GetConfigSection<UnicastBusConfig>();
if (unicastBusCfg != null)
{
    Hashtable assembliesToEndpoints = new Hashtable();
    Configure.TypesToScan.Where(t => typeof(IMessage).IsAssignableFrom(t)).ToList().ForEach(
            t => assembliesToEndpoints[t.Assembly.GetName().Name] = string.Empty
        );

    foreach (MessageEndpointMapping mapping in unicastBusCfg.MessageEndpointMappings)
        assembliesToEndpoints[mapping.Messages] = mapping.Endpoint;

    unicastCfg.ConfigureProperty(ub => ub.DistributorControlAddress, unicastBusCfg.DistributorControlAddress);
    unicastCfg.ConfigureProperty(ub => ub.DistributorDataAddress, unicastBusCfg.DistributorDataAddress);
    unicastCfg.ConfigureProperty(ub => ub.ForwardReceivedMessagesTo, unicastBusCfg.ForwardReceivedMessagesTo);
    unicastCfg.ConfigureProperty(ub => ub.MessageOwners, assembliesToEndpoints);
    unicastCfg.ConfigureProperty(ub => ub.ImpersonateSender, true);

    List<Type> handlers = new List<Type>();

    foreach (Type t in Configure.TypesToScan)
        if (IsMessageHandler(t))
        {
            config.Configurer.ConfigureComponent(t, ComponentCallModelEnum.Singlecall);
            handlers.Add(t);
        }

    unicastCfg.ConfigureProperty(ub => ub.MessageHandlerTypes, handlers);
}

Helper methods I used in this example

/// <summary>
/// Returns true if the given type is a message handler.
/// </summary>
/// <param name="t"></param>
/// <returns></returns>
public static bool IsMessageHandler(Type t)
{
    if (t.IsAbstract)
        return false;

    if (typeof(ISaga).IsAssignableFrom(t))
        return false;

    foreach (Type interfaceType in t.GetInterfaces())
    {
        Type messageType = GetMessageTypeFromMessageHandler(interfaceType);
        if (messageType != null)
            return true;
    }

    return false;
}

/// <summary>
/// Returns the message type handled by the given message handler type.
/// </summary>
/// <param name="t"></param>
/// <returns></returns>
public static Type GetMessageTypeFromMessageHandler(Type t)
{
    if (t.IsGenericType)
    {
        Type[] args = t.GetGenericArguments();
        if (args.Length != 1)
            return null;

        if (!typeof(IMessage).IsAssignableFrom(args[0]))
            return null;

        Type handlerType = typeof(IMessageHandler<>).MakeGenericType(args[0]);
        if (handlerType.IsAssignableFrom(t))
            return args[0];
    }

    return null;
}

Advertisement