pipeline

using System;
using System.Collections.Generic;
using System.Text;
using System.IO;
using System.Xml;
using System.ComponentModel;
using Microsoft.BizTalk.Component.Interop;
using Microsoft.BizTalk.Message.Interop;

using System.Drawing;
using System.Resources;
using System.Reflection;
using System.Diagnostics;
using System.Collections;
using System.ComponentModel.Design;
using Microsoft.BizTalk.Component;
using Microsoft.BizTalk.Messaging;


namespace MessageBatchPipelineCompoent
{
    [ComponentCategory(CategoryTypes.CATID_PipelineComponent)]
    [ComponentCategory(CategoryTypes.CATID_DisassemblingParser)]
    [System.Runtime.InteropServices.Guid("6118B8F0-8684-4ba2-87B4-8336D70BD4F7")]
    public class DisassemblePipeline : IBaseComponent,
        IDisassemblerComponent,
        IComponentUI,
        IPersistPropertyBag
    {
        //Used to hold disassembled messages
        private System.Collections.Queue qOutputMsgs = new System.Collections.Queue();
        private string systemPropertiesNamespace = @"http://schemas.microsoft.com/BizTalk/2003/system-properties";


        /// <summary>
        /// Batch size used to batch records
        /// </summary>
        private int _BatchSize;
        public int BatchSize
        {
            get { return _BatchSize; }
            set { _BatchSize = value; }
        }


        /// <summary>
        /// Default constructor
        /// </summary>
        public DisassemblePipeline()
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: Constructor"); // Debug
        }

        /// <summary>
        /// Description of pipeline
        /// </summary>
        public string Description
        {
            get
            {
                return "Component to batch (break) large message into multiple small messages";
            }
        }

        /// <summary>
        /// Name of pipeline
        /// </summary>
        public string Name
        {
            get
            {
                return "MessageBatchComponent";
            }
        }

        /// <summary>
        /// Pipeline version
        /// </summary>
        public string Version
        {
            get
            {
                return "1.0.0.10";
            }
        }


        /// <summary>
        /// Returns collecton of errors
        /// </summary>
        public System.Collections.IEnumerator Validate(object projectSystem)
        {
            return null;
        }

        /// <summary>
        /// Returns icon of pipeline
        /// </summary>
        public System.IntPtr Icon
        {
            get
            {
                return new System.IntPtr();
            }
        }


        /// <summary>
        /// Class GUID
        /// </summary>
        public void GetClassID(out Guid classID)
        {
            classID = new Guid("ACC3F15A-C389-4a5d-8F8E-2A951CDC4C19");
        }

        /// <summary>
        /// InitNew
        /// </summary>
        public void InitNew()
        {
        }


        /// <summary>
        /// Load property from property bag
        /// </summary>
        public virtual void Load(IPropertyBag propertyBag, int errorLog)
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: Load"); // Debug
            object val = null;
            try
            {
                val = ReadPropertyBag(propertyBag, "BatchSize");
                //propertyBag.Read("BatchSize", out val, 0);
            }
            catch (ArgumentException ex)
            {
                throw new ArgumentException("Error reading propertybag: " + ex.Message);
            }
            catch (Exception ex)
            {
                throw new ApplicationException("Error reading propertybag: " + ex.Message);
            }

            if (val != null)
                _BatchSize = (int)val;
            else
                _BatchSize = 1;
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: BatchSize: " + BatchSize.ToString()); // Debug

        }

        private object ReadPropertyBag(Microsoft.BizTalk.Component.Interop.IPropertyBag pb, string propName)
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: ReadPropertyBag, propName: " + propName); // Debug
            object val = null;
            try
            {
                pb.Read(propName, out val, 0);
            }
            catch (System.ArgumentException)
            {
                return val;
            }
            catch (System.Exception e)
            {
                throw new System.ApplicationException(e.Message);
            }
            return val;
        }

        /// <summary>
        /// Write property to property bag
        /// </summary>
        public virtual void Save(IPropertyBag propertyBag, bool clearDirty, bool saveAllProperties)
        {
            object val = (object)BatchSize;
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: Save: BatchSize " + val.ToString()); // Debug
            propertyBag.Write("BatchSize", ref val);
        }

        /// <summary>
        /// Disassembles (breaks) message into small messages as per batch size
        /// </summary>
        public void Disassemble(IPipelineContext pContext, IBaseMessage pInMsg)
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: Disassemble Start"); // Debug
            string originalDataString;
            try
            {
                //fetch original message
                Stream originalMessageStream = pInMsg.BodyPart.GetOriginalDataStream();
                byte[] bufferOriginalMessage = new byte[originalMessageStream.Length];
                originalMessageStream.Read(bufferOriginalMessage, 0, Convert.ToInt32(originalMessageStream.Length));
                originalDataString = System.Text.ASCIIEncoding.ASCII.GetString(bufferOriginalMessage);

                System.Diagnostics.Debugger.Log(0, null, "DEBUG: originalDataString: " + originalDataString); // Debug
            }
            catch (Exception ex)
            {
                throw new ApplicationException("Error in reading original message: " + ex.Message);
            }

            XmlDocument originalMessageDoc = new XmlDocument();
            StringBuilder messageString;
            try
            {
                //load original message
                originalMessageDoc.LoadXml(originalDataString);

                System.Diagnostics.Debugger.Log(0, null, "DEBUG: originalMessageDoc: " + originalMessageDoc.ToString()); // Debug

                //fetch namespace and root element
                string namespaceURI = originalMessageDoc.DocumentElement.NamespaceURI;
                string rootElement = originalMessageDoc.DocumentElement.Name;

                //start batching messages
                int counter = 0;

                messageString = new StringBuilder();
                messageString.Append("<" + rootElement + " xmlns:ns0='" + namespaceURI + "'>");
                foreach (XmlNode childNode in originalMessageDoc.DocumentElement.ChildNodes)
                {
                    counter = counter + 1;
                    System.Diagnostics.Debugger.Log(0, null, "DEBUG: counter: " + counter); // Debug
                    if (counter > BatchSize)
                    {
                        messageString.Append("</" + rootElement + ">");

                        //Queue message
                        CreateOutgoingMessage(pContext, messageString.ToString(), namespaceURI, rootElement);

                        counter = 1;
                        messageString.Remove(0, messageString.Length);
                        messageString.Append("<" + rootElement + " xmlns:ns0='" + namespaceURI + "'>");
                        messageString.Append(childNode.OuterXml);
                    }
                    else
                    {
                        messageString.Append(childNode.OuterXml);
                    }
                    System.Diagnostics.Debugger.Log(0, null, "DEBUG: foreach--messageString: " + messageString); // Debug
                }

                messageString.Append("</" + rootElement + ">");
                CreateOutgoingMessage(pContext, messageString.ToString(), namespaceURI, rootElement);
            }
            catch (Exception ex)
            {
                throw new ApplicationException("Error in writing outgoing messages: " + ex.Message);
            }
            finally
            {
                messageString = null;
                originalMessageDoc = null;
            }

        }

        /// <summary>
        /// Used to pass output messages`to next stage
        /// </summary>
        public IBaseMessage GetNext(IPipelineContext pContext)
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: GetNext"); // Debug
            if (qOutputMsgs.Count > 0)
                return (IBaseMessage)qOutputMsgs.Dequeue();
            else
                return null;
        }

        /// <summary>
        /// Queue outgoing messages
        /// </summary>
        private void CreateOutgoingMessage(IPipelineContext pContext, String messageString, string namespaceURI, string rootElement)
        {
            System.Diagnostics.Debugger.Log(0, null, "DEBUG: CreateOutgoingMessage"); // Debug
            IBaseMessage outMsg;

            try
            {

                System.Diagnostics.Debugger.Log(0, null, "DEBUG: messageString: " + messageString); // Debug

                //create outgoing message
                outMsg = pContext.GetMessageFactory().CreateMessage();
                outMsg.AddPart("Body", pContext.GetMessageFactory().CreateMessagePart(), true);
                outMsg.Context.Promote("MessageType", systemPropertiesNamespace, namespaceURI + "#" + rootElement.Replace("ns0:", ""));

                byte[] bufferOoutgoingMessage = System.Text.ASCIIEncoding.ASCII.GetBytes(messageString);
                outMsg.BodyPart.Data = new MemoryStream(bufferOoutgoingMessage);

                System.Diagnostics.Debugger.Log(0, null, "DEBUG: outMsg: " + outMsg.ToString()); // Debug

                qOutputMsgs.Enqueue(outMsg);
            }
            catch (Exception ex)
            {
                throw new ApplicationException("Error in queueing outgoing messages: " + ex.Message);
            }
        }


    }
}