BizTalk Messaging Archive Custom Solution

Lately one of the questions asked on BizTalk Server Forums intrigued me. The question was how one could archive a message including its message context.

Archiving received message
 
As soon as a message reaches BizTalk it can go through one of the default pipelines (XMLReceive,PassThruReceive) and a message context is added to incoming message.
                                                
The XMLReceive pipeline has a XmlDisassembler pipeline component on the disassembling stage.


Whenever an Xml message is received via the XmlReceive pipeline the XmlDisassembler will do the following tasks:

• Promote the "MessageType" context property by taking the combination of TargetNamespace and Root Element in the format of: Targetnamespace#RootElement. So one of context properties BizTalk will set (promote) is the MessageType.

Name: MessageType - Namespace: http://schemas.microsoft.com/BizTalk/2003/system-properties - http://BizTalk.Archiving.BankTransaction#Transaction

• Remove Envelopes and disassemble the interchanges
• Promote the content properties from interchange and individual document into message context based on the configured distinguished fields and promoted properties.

I will discuss here a custom disassembler pipeline component that will promote the “MessageType” context property and archive the message context and body to file. It will not perform the other two default actions by XmlDisassembler pipeline component.
Pipeline component is targeted for disassembler stage of the receive pipeline. The component category is CATID_DisassemblingParser will be set above the class amongst other attributes. Since the component is targeted at the disassembling stage the IDisassemblerComponent needs to be implemented. This interface has two methods, Disassemble and GetNext. In the Disassemble method you will find the implementation for archiving the received message.

        /// 

        /// Implements IDisassemblerComponent.Disassemble method.
        /// 
        /// 

Pipeline context
        /// 

Input message.
        /// Message
        /// 
        /// IComponent.Execute method is used to initiate
        /// the processing of the message in pipeline component.
        /// 
        public void Disassemble(IPipelineContext pContext, IBaseMessage pInMsg)
        {
            //Trace 
            System.Diagnostics.Debug.WriteLine("1.  Pipeline Disassemble Stage");

            //Create XmlDocument object
            XmlDocument xmlDoc = new XmlDocument();

            //Create a copy of the message
            IBaseMessage archiveMessage = pInMsg;

            //Trace
            System.Diagnostics.Debug.WriteLine("2.  Call GetMessagePayLoad()");

            //Get Message PayLoad
            xmlDoc = GetMessagePayLoad(archiveMessage);

            //Trace
            System.Diagnostics.Debug.WriteLine("3.  Message PayLoad :" + xmlDoc.OuterXml);

            // Promote MessageType in order to the Biztalk to have a unique key for evaluating the subscription
            archiveMessage.Context.Promote("MessageType", "http://schemas.microsoft.com/BizTalk/2003/system-properties", xmlDoc.DocumentElement.NamespaceURI + "#" + xmlDoc.DocumentElement.LocalName.ToString());

            //Debug 
            System.Diagnostics.Trace.WriteLine("4. Call ReadContextProperties");

            //Get the context properties and assign them to contextProperties  archiveMessage
            string contextProperties = ReadContextProperties(archiveMessage);

            //Debug 
            System.Diagnostics.Trace.WriteLine("5. Context Properties: " + contextProperties);

            //Get the message content (BodyPart)
            string messageBody = xmlDoc.OuterXml;

            //Debug 
            System.Diagnostics.Trace.WriteLine("6. Message Body: " + messageBody);

            //Debug 
            System.Diagnostics.Trace.WriteLine("7. Write to output file");

            //Write output
            using (StreamWriter outfile = new StreamWriter(_ArchiveLocation + System.Guid.NewGuid().ToString() + "_Message" + ".txt"))
            {

                //Debug 
                System.Diagnostics.Trace.WriteLine("8. File Location :" + _ArchiveLocation);

                outfile.Write(contextProperties + " " + Environment.NewLine + messageBody);

                //Debug 
                System.Diagnostics.Trace.WriteLine("9. Write to output file");
            }

            //Debug 
            System.Diagnostics.Trace.WriteLine("10. Pipeline Disassemble Stage Exit");


            //Return orginal message
            IBaseMessage outMessage;
            outMessage = pContext.GetMessageFactory().CreateMessage();
            outMessage.AddPart("Body", pContext.GetMessageFactory().CreateMessagePart(), true);
            IBaseMessagePart bodyPart = pInMsg.BodyPart;
            Stream originalStream = bodyPart.GetOriginalDataStream();
            originalStream.Position = 0;
            outMessage.BodyPart.Data = originalStream;

            outMessage.Context = PipelineUtil.CloneMessageContext(pInMsg.Context);

            _qOutMessages.Enqueue(outMessage);
            
            //Return orginal message to queue
            _qOutMessages.Enqueue(outMessage);
        }

        /// 

        /// Default method
        /// 
        /// 

Context
        /// null
        public IBaseMessage GetNext(IPipelineContext pContext)
        {
            if (_qOutMessages.Count > 0)
            {
                IBaseMessage msg = (IBaseMessage)_qOutMessages.Dequeue();
                return msg;
            }
            else
                return null;

        }

        /// 

        /// Read the context properties of the message
        /// 
        /// 

IBaseMessage archiveMessage
        /// string containing all context properties
        private string ReadContextProperties(IBaseMessage archiveMessage)
        {
            string name;
            string nmspace;
            string contextItems = "";

            for (int x = 0; x < archiveMessage.Context.CountProperties; x++)
            {
                archiveMessage.Context.ReadAt(x, out name, out nmspace);
                string value = archiveMessage.Context.Read(name, nmspace).ToString();
                contextItems += "Name: " + name + " - " + "Namespace: " + nmspace + " - " + value + "\r\n";
            }

            return contextItems;
        }

        /// 

        /// Method extract message into XMLDocument
        /// 
        /// 

IBaseMessage
        /// XML Document
        private XmlDocument GetMessagePayLoad(IBaseMessage archiveMessage)
        {
            IBaseMessagePart bodyPart = archiveMessage.BodyPart;
            Stream originalStream = bodyPart.GetOriginalDataStream();

            XmlDocument XMlDoc = new XmlDocument();
            XMlDoc.Load(originalStream);

            return XMlDoc; 
        }

The complete code can be found through MSDN Code Gallery here.

Buy versus Build

This is just a custom solution that took me a couple hours to develop and test. To add more functionality would mean more development work. A custom solution can bring a tremendous amount of flexibility and power. However it will also cost a fair amount of time to develop and some more time to maintain it (i.e. changes). An alternative can be buying one of the off-shelve products for archiving BizTalk messages like BizTalk Message Archiving Pipeline Component. This product has a great deal of features, offers support and has a license model.

BizTalk Tracking for Archiving

BizTalk offers tracking capabilities, which can be used to archive messages for a short period of time. Basically BizTalk tracking is created in such a way that you can use tracking for troubleshooting purposes not really for archiving. You can use tracking for archiving purposes, but then you need to be aware of fact that you have to configure the tracking and purging BizTalk database job correctly and move your tracking data! Why the job is so important is clearly explained in MSDN Archiving and purging the BizTalk Tracking Database:

As BizTalk Server processes more and more data on your system, the BizTalk Tracking (BizTalkDTADb) database continues to grow in size. Unchecked growth decreases system performance and may generate errors in the Tracking Data Decode Service (TDDS). In addition to general tracking data, tracked messages can also accumulate in the MessageBox database, causing poor disk performance….

By placing the backups somewhere else you can later on extract tracked messages from it. Tracked messages are compressed in BizTalk tracking database and you can extract these programmatically. Thiago Almeida has written post on how to do that. So again you need a custom solution to view the message body and its context.

You can use BizTalk tracking for archiving purposes, yet you need to offload the data from time to time to another another database. Retention on BizTalk databases is limited, because of the growth of data that eventually will impact BizTalk Server performance. In my personal opinion (view) I would not use BizTalk tracking for archiving purposes.

Archiving send message

In this post you have seen the implementation of the receive side of archiving a message. At send side a similar process can be performed to archive the message that is send by BizTalk to another system, application or service. Normally when a message is send by BizTalk, one of the default pipelines (XMLSend, PassThruSend) is used. With the XmlSend pipeline the reverse of what in a XmlReceive pipeline happens. The XMLSend has a XmlAssembler component in the Assemble stage. Whenever an Xml message is send via the XMLSend pipeline the XmlAssembler will do the following tasks:

• XML Assembler Builds envelopes as needed and appends XML messages within the envelope.
• Populates content properties on the message instance and envelopes. 

The custom assembler component will neither of these, it will only archive the message to file. Pipeline component is targeted for assembler stage of the send pipeline. The component category is CATID_AssemblingParser will be set above the class amongst other attributes.

[ComponentCategory(CategoryTypes.CATID_PipelineComponent)] [ComponentCategory(CategoryTypes.CATID_AssemblingSerializer)] [System.Runtime.InteropServices.Guid("728609D1-8282-4D73-B4D3-4792D6580537")]

Since the component is targeted assembling stage the IAssemblerComponent needs to be implemented. This interface has two methods, Assemble and AddDocument. In the AddDocument method you will find the implementation for archiving the send message.

        /// 


        /// Implements IAssemblerComponent.Assemble method.
        /// 

        /// 

Pipeline context
        /// Message
        /// 
        /// IComponent.Assemble method is used to
        /// 
        public IBaseMessage Assemble(IPipelineContext pipelineContext)
        {
            if (_qOutMessages.Count > 0)
            {
                IBaseMessage msg = (IBaseMessage)_qOutMessages.Dequeue();
                return msg;
            }
            else
                return null;
        }

        /// 

        /// IAssembler.AddDocument Method
        /// 
        /// 

Pipeline context
        /// 

Message send out
        public void AddDocument(IPipelineContext pContext, IBaseMessage pInMsg)
        {
            //Trace 
            System.Diagnostics.Debug.WriteLine("1.  Pipeline Assemble Stage");

            //Create XmlDocument object
            XmlDocument xmlDoc = new XmlDocument();

            //Create a copy of the message
            IBaseMessage archiveMessage = pInMsg;

            //Trace
            System.Diagnostics.Debug.WriteLine("2.  Call GetMessagePayLoad()");

            //Get Message PayLoad
            xmlDoc = GetMessagePayLoad(archiveMessage);

            //Trace
            System.Diagnostics.Debug.WriteLine("3.  Message PayLoad :" + xmlDoc.OuterXml);

            //Debug 
            System.Diagnostics.Trace.WriteLine("4. Call ReadContextProperties");

            //Get the context properties and assign them to contextProperties  archiveMessage
            string contextProperties = ReadContextProperties(archiveMessage);

            //Debug 
            System.Diagnostics.Trace.WriteLine("5. Context Properties: " + contextProperties);

            //Get the message content (BodyPart)
            string messageBody = xmlDoc.OuterXml;

            //Debug 
            System.Diagnostics.Trace.WriteLine("6. Message Body: " + messageBody);

            //Debug 
            System.Diagnostics.Trace.WriteLine("7. Write to output file");

            //Write output
            using (StreamWriter outfile = new StreamWriter(_ArchiveLocation + System.Guid.NewGuid().ToString() + "_Message" + ".txt"))
            {

                //Debug 
                System.Diagnostics.Trace.WriteLine("8. File Location :" + _ArchiveLocation);

                outfile.Write(contextProperties + " " + Environment.NewLine + messageBody);

                //Debug 
                System.Diagnostics.Trace.WriteLine("9. Write to output file");
            }

            //Debug 
            System.Diagnostics.Trace.WriteLine("10. Pipeline Disassemble Stage Exit");

            //Return orginal message
            IBaseMessage outMessage;
            outMessage = pContext.GetMessageFactory().CreateMessage();
            outMessage.AddPart ("Body", pContext.GetMessageFactory().CreateMessagePart(), true);
            IBaseMessagePart bodyPart = pInMsg.BodyPart;
            Stream originalStream = bodyPart.GetOriginalDataStream();
            originalStream.Position = 0;
            outMessage.BodyPart.Data = originalStream;

            outMessage.Context = PipelineUtil.CloneMessageContext(pInMsg.Context);
            
            _qOutMessages.Enqueue(outMessage); 
        }

The complete code can be found through MSDN Code Gallery here.

Conclusion

This is a very basic and straight forward custom solution to archive messages going in and out of BizTalk Server. It can be leveraged to create a more sophisticated archiving solution. There are however some considerations that have to be taken into account.

• When you expect a high volume of messages flowing in and out of BizTalk a lot of disk I/O will be the result when using these pipelines for archiving messages. So basically a high disk contention can occur when writing archived messages to same disk as where the BizTalk instance resides. A good approach would be having the archived messages stored on a separate dedicated disk.
• This solution shows writing archived messages to file, yet you can also choose to store them in a database or send the archived messages to a queue where they are picked up to be stored elsewhere. In the end you have to decide where you want your archived messages stored, for how long (retention) and possibly how to retrieve them if you want to look up a certain archived message later on. The latter will be rather difficult when having archived messages on file.
• Another thing you have to consider is data inside the messages. Is it data everyone can see or is sensitive data? Storing the files either on file or in database that easily accessible by others may not be a good option.

Having a solid, robust archiving solution in place for your BizTalk messages is an easy walk in the park. You’ll need a good design up front that is fit for purpose. I hope that with this post you have some food for thought.