NServiceBus Customization Part 1: IWantCustomInitialization

In this first part we're going to explore how to take control of NServiceBus during the initialization phase.  By implementing the IWantCustomInitialization interface, you take full control away from NSB Roles.  What is a Role?  A Role tells an Endpoint how to participate on the bus.  Typical Roles are the Server, Client, and Publisher Roles.  These are defined by a class in your project, typically named EndpointConfig.  A Role will define certain aspects of behavior, like whether or not it can receive messages(AsA_Client).

Most of the time you don't end up ever overriding the default behavior of the Roles.  When you do, you'll implement the IWantCustomInitialization interface and have full control over NServiceBus.  You will use the static NServiceBus.Configure class to get things going.  Let's take a look at all the options(as of 2.5, 3.0 follows):

• .With()/.WithWeb() - these first methods give you the ability to control assembly scanning.  When NSB fires up, it will scan the bin directory by default.  This gives you control over what gets scanned.  You can provide a list of Assemblies, a list of Types, or a path.  You'll want to use the WithWeb() version if you are using NSB with ASP.NET.  
• NServiceBus.AllAssemblies -  This class allows you to provide a list of assemblies not to load.  An example of this would be "AllAssemblies.Except( "Assembly1" ).And("Assembly2")".  
• .Synchronization() - tells NSB that message processing with occur in a synchronization domain.  This is useful for rich client applications where you need to marshal from a background thread back to the UI.
• .DefaultBuilder() or .MyContainerOfChoiceBuilder() - this next set of methods controls the container.  You can take the default container(AutoFac as of 2.5) or specify your own.  You'll need to include the builder implementation for your container as well as specify an instance of your container.  
• .UnicastBus() - use unicast messaging
• DoNotAutoSubscribe() - do not auto subscribe to a Publisher's messages
• ForwardReceivedMessagesTo( String ) - forward all messages to another endpoint.  This is helpful if you want to do some auditing.
• LoadMessageHandlers() - tells the bus to scan and load message handlers.  There are a couple interesting variants on this method:
• LoadMessageHandlers<TFirst> - tells the bus to load the handlers but load the assembly where TFirst resides before the rest
• LoadMessageHandlers<T>( First<T> ) - load the handlers and specifies that the handlers in the given order should go first.  To use the First class to specify ordering, it has 2 methods, "Then" and "AndThen".  A typical usages would be "First<IHandler1>.Then<IHandler2>().AndThen<IHandler3>()"
• PropagateReturnAddressOnSend( Boolean ) - receivers of the messages sent by this endpoint will see the address of the incoming messages
• .XmlSerializer() or .BinarySerializer() - specify the serialization technique you'd like to use.
• .CustomConfigurationSource( IConfigurationSource source ) - if you would rather not use the standard app.config file or its sections for Administrative configuration, you are welcome to use your own. Simply implement the IConfigurationSource interface and you are on your way.
• .MsmqTransport() - tells NSB to use MSMQ as the underlying communication mechanism.  This may be swapped out in favor of other transports.  Along with this setting, you can configure a few things underneath it:
• .DoNotCreateQueues() - don't try to create queues if they don't exist
• .IsolationLevel( IsolationLevel ) - the isolation level of the transaction that MSMQ uses.  You may consider tweaking this for performance reasons or otherwise.
• .IsTransactional( Boolean ) - for web scenarios, you may not want to use a transaction when pushing a message to the queue
• .PurgeOnStartup( Boolean ) - determine whether or not to purge queues when NSB starts up.
• .TransactionTimeout( Timespan ) - the time to wait for a transaction
• .MsmqSubscriptionStorage() or DBSubscriptionStorage() - tells NSB where to persist subscriptions.  If you are using Profiles, this is set in the Profile of choice(Lite, Integration, or Production).  If using MSMQ, the queue to use is in the config file.  If  using NHibernate, you can default to the config file for NH properties or supply an IDictionary to the overload.
• .DBSubscriptionStorageWithSQLLiteAndAutomaticSchemaGeneration() - pretty sure we all have this one down.
• .Sagas() - tells NSB to look for Sagas in the assemblies provided to NSB in the beginning of the process.
• .NHibernateSagaPersister() - this will use NH for Saga persistence.  Unless supplied to the overload, NSB will use the config file for the NH properties.  You can give this method an IDictionary of properties as this point.  
• NHibernateSagaPersisterWithSQLLiteAndAutomaticSchemaGeneration() - pretty self explanatory I think.
• .RijdaelEncryptionService() - if you are encrypting messages, this will load the keys from the config.
• .Log4Net() - there are a few overloads here for you to tweak Log4Net.  You have access to the Appender, AppenderSkeleton, and you and specify your own configuration.
• .RunCustomAction( Action ) - this is a place where you can run anything you want really.  Typically I've seen this used to tweak the container very early in the boot strapping process.

New 3.0 Options

     Some of the options have moved around to new classes, but the semantics remain the same.  Below are some of the new options available to address new features.

• .MessageForwardingInCaseOfFault() - tells the bus to forward messages when there is a fault
• .FtpTransport() - FTP transport options
• .ImpersonateSender(Boolean) - whether we should impersonate the sender or not
• .InMemoryFaultManagement() - sets up the in memory fault manager.  Faults are lost when the process shuts down.
• .NHibernateFaultManager() - sets up the NH fault manager.  Faults are saved to a database with the exception that caused them.
• .NHibernateUnitOfWork() - sets the UOW manager to use the internal NSB NH manager.  You can supply your own if you are also using NH in your code.

As we've found out, you have full control over how NSB is configured right from the start.  Next time we'll a close look at the Roles we talked about earlier in the article.

ReturnToSourceQueue PowerShell Script: Updated

UPDATE: I've added transaction support to the script.  Previous to this you would have lost a message if something had gone wrong.  PS does not support TransactionScope, so we had to use MessageQueueTransaction.

The ReturnToSourceQueue.exe tool that comes with NSB presented a challenge to us when we tried to deploy it on all our app servers.  We needed to schedule an on-demand job to run the tool by our Operations group.  Since the agent we use goes on each machine and is not cluster aware, this presented an issue.  Our resolution was to replicate the same functionality in PS with a bit of a twist.  We added the ability to hit a remote queue.  The script was done by one of our developers Brandon Moriarty and I've posted it to GitHub for everyone to enjoy!  Big thanks to Brandon for whipping this one up.  Don't forget I have a few other PS scripts in there to do some quick and dirty things with MSMQ.

NServiceBus Customization, Let me Count the Ways!

So as promised I'm going to start a new series on customizing the behaviour of NSB. I figured that we'd first try to count up all the ways to do it and then tackle each one by one. I'm going to categorize them by when they occur during the NSB lifetime, Startup, Processing, and Shutdown.  I'm not going to include everything(it would take forever), but I'm going to include the most popular places where you'd customize.

Startup

• IWantCustomInitialization - take over control of how the bus is initialized, skipping Roles
• Roles - AsA_* interfaces. Control how the bus is initialized.
• IWantToRunAtStartup - run a custom action once at startup
• IConfigureLoggingForProfile<T> - configure logging for a specific Profile
• Profiles - IProfile. Determine settings for a given target environment. OOTB includes "Lite", "Integration", and "Production"
• ISpecifyMessageOrderHandling - creates a "pipeline" of messages and gives them a specific order
• The Container - you can pretty much change anything in the container, including passing in your own that is loaded with your stuff.

Processing

• IMessageMutator - manipulate inbound and outbound messages
• IMessageModule - inject some logic before and after message processing
• IManageUnitsOfWork - UOW pattern handling
• IManageMessageFailures - custom failed message handling
• Processing Events - StartedMessageProcessing, TransportMessageReceived, FinishedMessageProcessing, FailedMessageProcessing.  You can use these to do something when message events are happening(say perf counters)

Shutdown

• IWantToRunAtStartup - there is a Stop method here that gets called once during shutdown

ReturnToSourceQueue PowerShell Script

The ReturnToSourceQueue.exe tool that comes with NSB presented a challenge to us when we tried to deploy it on all our app servers.  We needed to schedule an on-demand job to run the tool by our Operations group.  Since the agent we use goes on each machine and is not cluster aware, this presented an issue.  Our resolution was to replicate the same functionality in PS with a bit of a twist.  We added the ability to hit a remote queue.  The script was done by one of our developers Brandon Moriarty and I've posted it to GitHub for everyone to enjoy!  Big thanks to Brandon for whipping this one up.  Don't forget I have a few other PS scripts in there to do some quick and dirty things with MSMQ.

Oracle AQS Transport Now on NServiceBus-Contrib

Source has been committed to the v2.5 branch on GitHub. I'll start work on the master(a.k.a. 3.0 version) shortly.

Oracle AQS Transport Complete & the Next Article Series

This went together pretty quickly. I have a local version tested and working from both directions. I can push messages in from the database side via a stored procedure and also from the .NET world. All that is left is to fork NServiceBus-Contrib and get the code into that solution structure and figure out how to build it. It's using Ruby/rake which I've never used before, but the format seems simple enough.

After that work is done, it's back to getting the 3.0 branch to function. I'll need to work on the serialization issues with the headers, hopefully that is not a big deal. I'll be sure to include all the SQL scripts I used to setup and test the transport.

During our implementation of NSB, we came across many times where we had to customize NSB to do various things. I'm thinking the next series will be on all the ways you can customize and extend NSB.

Working on Oracle AQS Transport

Internally we have a need to hook into certain changes within some data that is housed in Oracle. Our solution will be to use Oracle CDC(Change Data Capture) to grab the info we need right from the redo logs and push that data into a table. We're going use the table to provide those who would like to do ETL a simple way to grab data that is not on the live tables and only contains the changes. From this table we'll push the rows into Oracle AQ(Advanced Queuing). From here we have the opportunity to drive that data right onto NSB much like the Service Broker is being used in this article.

Thanks to the contributors over at NServiceBus/NServiceBus-Contrib I have a great head start. I started out going down the NSB 3.0 route since the barrier to entry has been lowered greatly when attempting to create a custom Transport. Inadvertently I realized that I'm going to need to support NSB 2.5 seeing as how most of my shop is still on 2.0 and moving to 2.5, not only that 3.0 is still considered alpha or pre-alpha even. I also ran into an interesting serialization error in 3.0 anyway, once which I'll have to revisit later. The message headers on TransportMessage in 3.0 is now a Dictionary(used to be a List) which the serializer does not like one bit.

Luckily I was able to re-use the code from the 3.0 implementation. Going back to the 2.5 methodology has been a bit painful, and I'm glad that the custom Transport story is changing drastically. I have everything all set up and I'm ready to test. I'm even running Oracle 10g Express locally so this thing should be easy to try out for everyone. I'm interested to see if I can clean up some of the serialization work from the Service Broker implementation. Oracle AQ stores everything in a special table that backs the queue. You can tell it to make the column that stores the data to be an XMLType column or RAW, which I'm hoping plays more nicely with the serializers.

Once I get something more solid my plan is to put the code out on my personal repository in github and then work on getting it integrated into NServiceBus-Contrib. Let me know if anyone has already built this or is trying to, I'm always on the lookout for any problems or suggestions that have come up.