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.

NServiceBus 3.0: Message Mutators

In NSB 2 it was a bit more tricky to change messages as they were sent to and from endpoints.  In NSB 3 we now have a couple of simple interfaces to plug in our custom logic to change messages.  The typical scenario this is used for is to encrypt all or part of a message.  The encryption message mutator is built right in and can be used at any time.  In this article we'll show how to create a custom message mutator to simply do some math on Double fields in a message.  Let's first take a quick look at the interfaces involved.

Each interface gives us access to the message so that we can mutate on the inbound and/or outbound message.  All we have to do as a consumer is implement the desired interface and load it into the NSB container.  For my mutator, I will implement IMessageMutator and only implement the outgoing message for the time being.

public class MultiplierMutator : IMessageMutator
    {
        #region IMutateOutgoingMessages Members

        public NServiceBus.IMessage MutateOutgoing(NServiceBus.IMessage message)
        {
            var doubles = message.GetType().GetProperties().Where(p => typeof(double).IsAssignableFrom(p.PropertyType));

            foreach (var dbl in doubles)
            {
                double dblValue = (double)dbl.GetValue(message, null);
                dbl.SetValue(message, dblValue * 5, null);
            }

            return message;
        }

        #endregion

        #region IMutateIncomingMessages Members

        public NServiceBus.IMessage MutateIncoming(NServiceBus.IMessage message)
        {
            return message;
        }

        #endregion
    }

In this class we first find all Double fields. Once we have that list we simply multiply each field by an arbitrary number, in this case 5. The last thing we need to do is let the container know about the mutator. We do this in our implementation of IWantCustomInitialization.

NServiceBus.Configure.Instance.Configurer.ConfigureComponent<MultiplierMutator>(NServiceBus.ObjectBuilder.DependencyLifecycle.InstancePerCall);

Now we can fire up our endpoints and watch what happens. I stubbed this into the Encryption sample that comes with the NSB download. In the Client project I added a value to the Result field on the message.

while (Console.ReadLine() != null)
            {
                Bus.Send<MessageWithSecretData>(m =>
                    {
                        m.Secret = "betcha can't guess my secret";
                        m.Result = 5;
                    });
            }

Now for the output from the Server project:

So that's it! Now we can dream up any kind of change to our messages. Maybe you want to compress or hash some of the messages, it's all up to you.

MSMQ PowerShell Scripts

I thought I'd hand out a couple of PowerShell scripts that I whipped up for our Server Admin friends.  You can use these to do some basic things with MSMQ such as send a message, browse messages in a queue, and return messages to a source queue.  Check them out: https://github.com/afyles/Blog/tree/master/PowerShell

Clustered MSDTC Mutual Authentication Gotcha Between 2003/2008 Servers

This is more of a mental note that anything else, but we ran across an issue with configuring our subscriptions database on a cluster.  What we had set up was a 2008 cluster with our Publisher on it and a 2003 cluster with our subscriptions DB on it.  MSTC was giving us a very distinct error:

[SQL: SELECT this_.SubscriberEndpoint as y0_ FROM [Subscription] this_ WHERE this_.MessageType in (?, ?, ?)]; Communication with the underlying transaction manager has failed.; The MSDTC transaction manager was unable to push the transaction to the destination transaction manager due to communication problems. Possible causes are: a firewall is present and it doesn't have an exception for the MSDTC process, the two machines cannot find each other by their NetBIOS names, or the support for network transactions is not enabled for one of the two transaction managers. (Exception from HRESULT: 0x8004D02A)

The solution for this was that 2003 servers in cluster doesn't support Mutual Authentication.  We ended up just using the Incoming Caller authentication between the clusters and everything started working again.  Hope this helps someone else with the same problem.

NServiceBus 3.0: Alternative Transports

Back in NSB 2.0 you could create your own transport but it took a bit of work.  In 3.0 the concept has been better abstracted to allow us to write custom transports more easily.  What was introduced was a new set of interfaces for us to implement:

These simple interfaces have been plugged into the transport layer and makes it pretty painless to derive a new transport.  In  NSB 3.0, we already have an FTP transport and an Azure transport.  I'm going to run through the FTP sample.  First things first, you have to setup a couple of FTP sites on your machine.  I followed the directions here.  To get the sample to work you have to set up one on port 1090 and one on port 1091.

Next we need to mod the config to match our local one, here is how my config looked for the Client:

Here is the config for the Server:

Now we can fire the whole thing up and start sending messages. Once a message is sent you can see it in your FTP directory.

Each is a file with the serialized data.  Its as simple as that!  It will be interesting to see what kind of other transports people plug in.  I'm hoping that there will be some other queue based transports like MQ Series.

NServiceBus 3.0: Fault Management

In NSB 3.0 we get some better support for message faults. Prior to 3.0 we were able to capture messages that faulted after a configurable number of retries. The difficulty with this is that since the actual exception was not captured with the fault, we as developers weren't able to do to much with that message besides blindly replay the message. In 3.0 this all changes in that there is a dedicated set of classes for fault management. Below is a class diagram of the hierarchy.

Out of the box we get 3 ways to manage faults.  The Forwarder simply does what NSB has done in the past, which is move the message to a designated queue.  The InMemory manager simply holds the message in memory for the lifetime of the host process.  Lastly, the NHibernate Fault Manager allows us to store the message along with the exception that caused it to be faulted.  This allows us to be smarter about how we handle faults.

To follow along you will need to download the 3.0 branch from GitHub and build it.  From there open up the "FaultHandling" sample.  When I downloaded this sample, it did not work right away.  If you fire up the solution, nothing really happens.   First and foremost you have to change the MyClient project endpoint to be transactional.  If it is not, the fault management gets skipped right over in code. Simply edit the EndpointConfig.cs file to configure it AsA_Server.

public class EndpointConfig : IConfigureThisEndpoint, AsA_Server {}

The next issue I ran into is that the default profile(Lite) gives us the InMemory Fault Manager. The sample configured this via the IWantCustomInitialization insertion point. The only way I managed to get this to work was to move it to a custom profile, aptly named "Custom"

class Custom : IProfile{}

  class CustomProfile : IHandleProfile<Custom>
  {
        #region IHandleProfile Members

        public void ProfileActivated()
        {
            Configure.Instance.InMemorySagaPersister();
            Configure.Instance.NHibernateFaultManagerWithSQLiteAndAutomaticSchemaGeneration();
        }

        #endregion

  }

Now that we have everything configured appropriately we can run MyClient. After launching the solution, simply enter an "s" to send a few messages. Sooner or later one of them will fail and start the retry cycle. Once that is complete, the NH Fault Manager kicks in and pushes the messages to SQLLite in my case. Here is what the output looks like:

Next we can dive to SQLLite to see what the output looks like.  Note that the Message column hasn't been populated, but I'm hoping this simply a SQLLite issue.

From here we can now make better decisions on what to do with these faulted messages.  I'm going to try using a full fledged SQL Server instance to see if the "Message" column gets populated.  If not, looks like I'll be reporting a bug.