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.