Calling Pipelines with custom property settings from an orchestration

In my previous blog post I explained how to execute pipelines from an orchestration using the default pipeline property settings.
Now I would like to show you how to call pipelines from an orchestration while configuring the pipeline with custom property settings. To achieve this the standard methods wouldn't suffice.

First of all you'll need to add some extra functionalities. Here's my code to achieve this:

using System;
using System.Collections.Generic;
using System.Text;
using System.Xml;
using System.Data.SqlClient;
using System.Runtime.Serialization;

using Microsoft.XLANGs.Pipeline;
using Microsoft.BizTalk.PipelineOM;
using Microsoft.XLANGs.BaseTypes;
using System.IO;
using System.Reflection;
using System.Collections;
using System.Data;
using System.Text.RegularExpressions;
using Microsoft.XLANGs.RuntimeTypes;
using Microsoft.XLANGs.Core;
using MIME;
using System.Net;

namespace TP.BTS.GlobalFunctions
{

public class PipelineInstance
    {
        internal Microsoft.BizTalk.PipelineOM.Pipeline pipeline = null;

        public static PipelineInstance CreateInstance(Type type)
        {
            PipelineInstance instance = new PipelineInstance();
            instance.pipeline = (Pipeline)Activator.CreateInstance(type);
            return instance;
        }

        public void ApplyInstanceConfig(String config)
        {
            try
            {
                using (Stream stream = new MemoryStream(Encoding.Unicode.GetBytes(config)))
                    LoadPerInstanceConfig(pipeline, stream);
            }
            catch (Exception e)
            {
                System.Diagnostics.Debug.WriteLine(String.Format("{0}", e));
                throw;
            }
        }

        #region Implementation

        private static void LoadPerInstanceConfig(Pipeline pipeline, Stream stream)
        {
            try
            {
                MethodInfo method =
                    typeof(Pipeline).GetMethod(
                        "LoadPerInstanceConfig"
                        , BindingFlags.NonPublic | BindingFlags.Instance);
                method.Invoke(pipeline, new object[] { stream });
            }
            catch (System.Exception e)
            {
                System.Diagnostics.Debug.WriteLine(e);
                throw;
            }
        }

        #endregion
    }

    public static class PipelineManager
    {
        #region receive pipeline
        public static ReceivePipelineOutputMessages ExecuteReceivePipeline(Type type, XLANGMessage message)
        {
            string EmptyConfig =
                        @"<Root
                        xmlns:xsd='http://www.w3.org/2001/XMLSchema'
                        xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'>
                        <Stages />
                    </Root>";

            return ExecuteReceivePipeline(type, EmptyConfig, message);
        }

        public static ReceivePipelineOutputMessages ExecuteReceivePipeline(Type type, String config, XLANGMessage message)
        {
            try
            {
                PipelineInstance pipeline = PipelineInstance.CreateInstance(type);
                pipeline.ApplyInstanceConfig(config);
                return ExecuteReceivePipeline(pipeline, message);
            }
            catch (TargetInvocationException e)
            {
                System.Diagnostics.Debug.WriteLine(e);
                ThrowPipelineException(e);
                throw;
            }
            catch (XLANGPipelineManagerException e)
            {
                throw;
            }
            catch (Exception e)
            {
                System.Diagnostics.Debug.Assert(false, String.Format("Unexpected exception: {0}", e));
                System.Diagnostics.Debug.WriteLine(e);
                throw;
            }
        }

        public static ReceivePipelineOutputMessages ExecuteReceivePipeline(PipelineInstance pipeline, XLANGMessage message)
        {
            Microsoft.BizTalk.PipelineOM.Pipeline pipelineOM = pipeline.pipeline;

            try
            {
                MethodInfo method =
                    typeof(XLANGPipelineManager).GetMethod(
                        "ExecutePipeline"
                        , BindingFlags.NonPublic | BindingFlags.Static
                        , null, new Type[] { typeof(ReceivePipeline), typeof(XLANGMessage) }
                        , null);
                return (ReceivePipelineOutputMessages)method.Invoke(null, new object[] { pipelineOM, message });
            }
            catch (TargetInvocationException e)
            {
                System.Diagnostics.Debug.WriteLine(e);
                ThrowPipelineException(e);
                throw;
            }
        }
        #endregion

        #region send pipeline
        public static void ExecuteSendPipeline(Type type, SendPipelineInputMessages pipelineinputmsg, XLANGMessage message)
        {
            string EmptyConfig =
                        @"<Root
                        xmlns:xsd='http://www.w3.org/2001/XMLSchema'
                        xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'>
                        <Stages />
                    </Root>";

            ExecuteSendPipeline(type, EmptyConfig, pipelineinputmsg, message);
        }

        public static void ExecuteSendPipeline(Type type, String config, SendPipelineInputMessages pipelineinputmsg, XLANGMessage message)
        {
            try
            {
                PipelineInstance pipeline = PipelineInstance.CreateInstance(type);
                pipeline.ApplyInstanceConfig(config);
                ExecuteSendPipeline(pipeline, pipelineinputmsg, message);
            }
            catch (TargetInvocationException e)
            {
                System.Diagnostics.Debug.WriteLine(e);
                ThrowPipelineException(e);
                throw;
            }
            catch (XLANGPipelineManagerException e)
            {
                throw;
            }
            catch (Exception e)
            {
                System.Diagnostics.Debug.Assert(false, String.Format("Unexpected exception: {0}", e));
                System.Diagnostics.Debug.WriteLine(e);
                throw;
            }
        }

        public static void ExecuteSendPipeline(PipelineInstance pipeline, SendPipelineInputMessages pipelineinputmsg, XLANGMessage message)
        {
            Microsoft.BizTalk.PipelineOM.Pipeline pipelineOM = pipeline.pipeline;

            try
            {
                MethodInfo method =
                    typeof(XLANGPipelineManager).GetMethod(
                        "ExecutePipeline"
                        , BindingFlags.NonPublic | BindingFlags.Static
                        , null, new Type[] { typeof(SendPipeline), typeof(SendPipelineInputMessages), typeof(XLANGMessage) }
                        , null);
                method.Invoke(null, new object[] { pipelineOM, pipelineinputmsg, message });
            }
            catch (TargetInvocationException e)
            {
                System.Diagnostics.Debug.WriteLine(e);
                ThrowPipelineException(e);
                throw;
            }
        }
       
        #endregion

        #region Implementation

        private static void ThrowPipelineException(Exception innerException)
        {
            ConstructorInfo constructor =
                typeof(XLANGPipelineManagerException).GetConstructor(
                    BindingFlags.NonPublic | BindingFlags.Instance
                    , null
                    , new Type[] { typeof(String), typeof(Exception) }
                    , null);
            throw (XLANGPipelineManagerException)constructor.Invoke(
                    new object[] {
                    innerException != null ? 
                    innerException.InnerException.Message != null? 
                    innerException.InnerException.Message : innerException.Message: 
                    String.Empty, innerException });
        }

        #endregion
    }
    }

Now, here's how to execute a receive pipeline using the custom property settings.
First create a variable of type TP.BTS.GlobalFunctions.PipelineInstance (called ReceivePipeline in my example). Also create a variable PipelineOutputMsg of type Microsoft.XLANGs.Pipeline.ReceivePipelineOutputMessages.
This would be the code to execute the pipeline then:

ReceivePipeline = 
TP.BTS.GlobalFunctions.PipelineInstance.CreateInstance(typeof(TC_FFRECEIVE.TC_FFRECEIVE));

ReceivePipeline.ApplyInstanceConfig(
@"<Root xmlns:xsd='http://www.w3.org/2001/XMLSchema' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'>
  <Stages>
    <Stage CategoryId='9d0e4103-4cce-4536-83fa-4a5040674ad6'>
      <Components>
        <Component Name='TP.BTS.PIPELINES.CustomXMLRECEIVECOMPONENT'>
          <Properties />
        </Component>
      </Components>
    </Stage>
    <Stage CategoryId='9d0e4105-4cce-4536-83fa-4a5040674ad6'>
      <Components>
        <Component Name='Microsoft.BizTalk.Component.FFDasmComp'>
          <Properties>
            <DocumentSpecName vt='8'>FLATFILE.ORDERS.FF_ORDERS,FLATFILE.ORDERS, Version=1.0.0.0, Culture=neutral, PublicKeyToken=a8e07cee919011d4</DocumentSpecName>
          </Properties>
        </Component>
      </Components>
    </Stage>
  </Stages>
</Root>");



PipelineOutputMsg = TP.BTS.GlobalFunctions.PipelineManager.ExecuteReceivePipeline(ReceivePipeline,MsgICMTxt);

The XML assigned when calling the ApplyInstanceConfig is the same as you can find when exporting the binding file with the Biztalk Admin Console.
It is that simple :).

Calling a send pipeline with custom property settings would be done like this:
Create a variable SendPipelineInput from type Microsoft.XLANGs.Pipeline.SendPipelineInputMessages. I've also used a variable PipelineConfig of type System.String.
Here's the code I've used:

SendPipelineInput.Add(MsgINVOIC_EDI);

PipelineConfig = @"<Root xmlns:xsd='http://www.w3.org/2001/XMLSchema' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'>
  <Stages>
    <Stage CategoryId='9d0e4101-4cce-4536-83fa-4a5040674ad6'>
      <Components>
        <Component Name='TP.BTS.PIPELINES.CUSTOMSENDEDIPARTYRESOLUTION.CUSTOMSENDEDIPARTYRESOLUTION'>
          <Properties>
            <OverrideRcvprn vt='8'>" + MsgINVOIC_GEN(ORDERS.ORDERSPropertySchema.RCVPRN) + @"</OverrideRcvprn>
            <OverrideRecipient vt='11'>-1</OverrideRecipient>
            <OverrideSndprn vt='8'>" + MsgINVOIC_GEN(ORDERS.ORDERSPropertySchema.SNDPRN) + @"</OverrideSndprn>
          </Properties>
        </Component>
      </Components>
    </Stage>
    <Stage CategoryId='9d0e4107-4cce-4536-83fa-4a5040674ad6'>
      <Components>
        <Component Name='Microsoft.BizTalk.Edi.Pipelines.EdiAssembler'>
          <Properties />
        </Component>
      </Components>
    </Stage>
    <Stage CategoryId='9d0e4108-4cce-4536-83fa-4a5040674ad6'>
      <Components>
        <Component Name='TP.BTS.PIPELINES.CustomXMLSENDCOMPONENT'>
          <Properties />
        </Component>
      </Components>
    </Stage>
  </Stages>
</Root>";

OrdersOutputInterchange = null;

TP.BTS.GlobalFunctions.PipelineManager.ExecuteSendPipeline(typeof
(CUSTOM_EDI_SEND.CUSTOM_EDI_SEND), PipelineConfig, SendPipelineInput,OrdersOutputInterchange);

And that should do the trick.

Hope you enjoyed this post.