Starting BizTalk Server 2004 orchestrations using the MessageAgent.
Jesús Rodríguez.
Didiosky Benítez.
Introduction.
The messaging system of BizTalk Server 2004 relies over a powerful publish-subscribe architecture that uses Sql Server as primary store. The main component to interact directly with this system is the MessageAgent. The main idea of this article is to show how to uses the MessageAgent to start BizTalk orchestrations from .Net code.
MessageBox and the MessageAgent.
It is not an objective of this article to explain the BizTalk Server 2004 subscription mechanisms. About this topic Lee Graber post two interesting comments (Is there a Pub/Sub System Underneath BizTalk? and Okay, so Where do Subscriptions Come From?) that go deep inside this pub-sub architecture. This section however gives a briefly explanation of basic components in the BizTalk Server 2004 subscriptions system.
The BizTalk Server 2004 pub-sub system can be segmented in two main parts. The first part is the storage portion and is known as the MessageBox. This part provides the storage mediums and the primary logic to maintain the state of a subscription. The second part is the low-level API that the different BizTalk Server 2004 engines uses to interact with the MessageBox, this part is known as the MessageAgent.
The main idea behind BizTalk Server 2004 subscriptions is amazingly simple. Orchestrations are the principal subscribers via its receive actions. There are two main types of subscriptions (see Lee post) activation and correlation subscriptions. In this article we primarily focus in Instance subscriptions that always creates a new instance of a service. An activation subscription is created when you enlist a service, and immediately the subscription enters in a stopped state. In the time that the subscription remains stopped all the messages that match this subscription go to the “SuspendedQ”. When the service starts the subscription pass to the active state and the messages goes to the “WorkQ”. When you perform an “unenlist” of your services you simply destroy the subscriptions associated with those services. This simply mechanism makes possible to keep the state of the subscriptions and its associated messages.
As I pointed early, the MessageBox contains the primary logic (stored procedures, functions, triggers) and the storage representation (tables, views) to interact with subscriptions. A subscription is formed by a set of predicates that are evaluated against each incoming messages to determine is that message match with the subscription. Every message contains a series of contextual properties that are used by those predicates. When a message arrives to the MessageBox the bts_InsertProperty sp is called to insert the each contextual property in the MessageProps table. Then the bts_FindSubscriptions sp is executed to find the available subscriptions and if any subscription is founded the bts_InsertMessage is called to insert the message into the MessageBox. The predicates that forms the subscription are stored in the following tables:
- Bitwise AND
- Equals
- Exists
- GreaterThanOrEquals
- GreaterThan
- LessThanOrEquals
- LessThan
- NotEquals
It is obvious that the BizTalk engines needs a via to interact to the MessageBox with a different level of abstractions. That is precisely the MessageAgent role; it provides an object hierarchy that makes possible to interact with the MessageBox. The MessageAgent has its own message definition; a message has context properties and is formed by a part that contains the data. The message contents are absolutely irrelevant to the MessageAgent and would be defined by the superior tiers. There are to main types of properties: context written and promoted. The main difference is that promoted property is used against the subscriptions predicates but both types of properties are stored in the MessageBox.
GetObject Subscription.
BizTalk Server 2004 orchestrations can be activated in different ways. One of the most typical ways occurs when some messages arrives to a well-defined location using some transport protocol. However an orchestration can start another orchestration passing a set of parameters. This is the role performed by the Start Orchestration shape in the orchestration designer. A very recurrent question in BizTalk developers is: can we start an orchestration from .NET code passing the required parameters? Well the answer is yes and (surprise, surprise) is solved using subscriptions.
To invoke an orchestration we can use a special type of subscription that is created in the MessageBox and is know as GetObject subscription. The invoker orchestration calls the ExecuteService method of the Microsoft.BizTalk.XLANGs.BTXEngine.BTXService class to pass a well-defined message and the required parameters that activates the invoked orchestration. The Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage class abstracts this message type. In general terms all we must do to execute an orchestration is submit a message to the MessageBox that belong to the ExecMessage type and specify the orchestration that should be activated. These steps are absolutely enough to execute an orchestration that receives no parameters. In the next section we explore the basic code to submit messages to the MessageBox that starts non-parameterized and parameterized orchestrations.
Starting orchestrations using the MessageAgent.
To interact with the MessageBox we use the MessageAgent, which in managed environments is abstracted in Microsoft.BizTalk.Agent.Interop namespace.
To initialize the MessageAgent we must perform a series of basic steps that are shown in the next code section.
| IBTMessageAgent agent= new BTMessageAgent() as IBTMessageAgent; agent.SetConfigDbLocation(MyServer…, My Configuration DB…); IBTMessageSpace sp= agent.RegisterAsPublisher(My Bts Group…, Bts Application name…); IBTMessageSpaceEx spx= (IBTMessageSpaceEx)sp; Guid spuid= Guid.Empty; Guid agentid= BTXProperties.MessageAgentClassID; agent.RegisterService(ref agentid, null); IBTMessageBatch expex= sp.GetMessageBatch(ref spuid); |
The code is very simple. We first create an agent and set the configuration information that includes the SQL Server name and the BizTalk management database. The following steps register the agent as publisher and obtain the message batch that will be used to submit the messages.
To start an orchestration that receive no parameters we add a few lines that deals directly with the message composition.
| IBTMessageAgent agent= new BTMessageAgent() as IBTMessageAgent; agent.SetConfigDbLocation(MyServer…, My Configuration DB…); IBTMessageSpace sp= agent.RegisterAsPublisher(My Bts Group…, Bts Application name…); IBTMessageSpaceEx spx= (IBTMessageSpaceEx)sp; Guid spuid= Guid.Empty; Guid agentid= BTXProperties.MessageAgentClassID; agent.RegisterService(ref agentid, null); IBTMessageBatch expex= sp.GetMessageBatch(ref spuid); //new code… IBTMessageAgentFactory msgfactory = (IBTMessageAgentFactory)agent; //creates the message... IBTMessage msg= (IBTMessage)msgfactory.CreateMessage(); msg.Context.Write("MessageType", "http://schemas.microsoft.com/BizTalk/2003/system-properties", "Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage"); msg.Context.Promote("MessageType", "http://schemas.microsoft.com/BizTalk/2003/system-properties", "Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage"); msg.Context.Write("ActivationServiceID", "http://schemas.microsoft.com/BizTalk/2003/system-properties", My Orchestration full name...); msg.Context.Promote("ActivationServiceID", "http://schemas.microsoft.com/BizTalk/2003/system-properties", My Orchestration full name...); msg.OriginatorSID= My user...; msg.OriginatorSID= My PID…; msg.IgnoreRoutingFailures= true; IBTMessage[] msgs= new IBTMessage[1]{msg}; expex.PostMessages(msgs, 1); //gets a transaction... Type TXType = Type.GetTypeFromCLSID(new Guid("30C1F639-A3A6-4E0A-879D-0B15020754C6")); object TXObj = Activator.CreateInstance(TXType); IXLANGTranDispImpl TXImpl = (IXLANGTranDispImpl) TXObj; ITransaction CurrentTX= (ITransaction)TXImpl.GetTransaction(Guid.NewGuid().ToString(), -1, 1048576); IBTOperationStatus opst= null; try { expex.CommitBatch(CurrentTX, CommitBatchFlags.CB_Default, out opst); txxx.Commit(0, 1, 0); } catch(Exception ex) { Marshal.ReleaseComObject(opst); } |
The code basically creates a new message using a Message Factory and set the MessageType and ActivationServiceID contextual properties that are evaluated against the predicates to find the available subscriptions. The final steps commit the message batch using a previously initialized transaction.
The generic code explained above works well with orchestration that receive 0 parameters. But you agree with me in that is not the most typical case. It is much more common that orchestrations receive some type of parameters as part of its activation process. The MessageAgent have no notion about parameters, it represents the parameters as a non-body MessagePart. In the case of GetObject subscription we must add a MessagePart for each parameter and one special part, known as MetaData, that describes the rest of the parameters. The MetaData is used by the superior tiers to deserialize the parameters and is required in most of the cases.
In object terms the MetaData part is typically an object of type Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamDataArray, this is a public class inside the internal ExecMesage class. This class contains an array of Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamData that encapsulates the MetaData associated with a parameter. The stream of data that the MessageAgent send to the MessageBox is the result of serializes an instance of ParamDataArray using the BinaryFormatter. Suppose that we have an orchestration that receives two string parameters an output the result of its concatenation. The next code makes an exhaustive use of reflection to obtain the binary stream that contains the metadata of the two string parameters.
| //constants section… #region consts const string gacassb= "Microsoft.XLANGs.BizTalk.Engine, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35"; const string iFirstPropName= "iFirst"; const string nPartsPropName= "nParts"; const string TypePropName = "Type"; const string ArrayPropName= "Item"; #endregion //this method returns the metadata relative to the two string parameters… public Stream GetValueParamTestMetaData() { Assembly btxassb= Assembly.Load(gacassb); Type ParamArrayType= btxassb.GetType("Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamDataArray"); Type ParamDataType= btxassb.GetType("Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamData"); object[] args= new object[]{2}; object ParamsMetaData= Activator.CreateInstance(ParamArrayType, args); for(int index= 0; index<= 1; index++) { //creates the first param... object ParamMetaData= Activator.CreateInstance(ParamDataType); //set the metadata properties... ParamDataType.GetProperty(iFirstPropName).SetValue(ParamMetaData, index + 1, null); ParamDataType.GetProperty(nPartsPropName).SetValue(ParamMetaData, 1, null); ParamDataType.GetProperty(TypePropName).SetValue(ParamMetaData, typeof(System.String), null); //creates the param list... ParamArrayType.GetProperty(ArrayPropName).SetValue(ParamsMetaData, ParamMetaData, new object[]{index}); } //serializes the meta data... BinaryFormatter formatter= new BinaryFormatter(); MemoryStream buffer= new MemoryStream(); formatter.Serialize(buffer, ParamsMetaData); buffer.Seek(0, SeekOrigin.Begin); return buffer; } |
Now we make a little modification to our base code to include the parameters and metadata MessageParts.
| //this method serializes gets the data stream of the parameter… public Stream GetValueParamTestData(string ParamValue) { MemoryStream buffer= new MemoryStream(); BinaryFormatter formatter= new BinaryFormatter(); formatter.Serialize(buffer, ParamValue); buffer.Seek(0, SeekOrigin.Begin); return buffer; } //the general code… IBTMessageAgent agent= new BTMessageAgent() as IBTMessageAgent; agent.SetConfigDbLocation(MyServer…, My Configuration DB…); IBTMessageSpace sp= agent.RegisterAsPublisher(My Bts Group…, Bts Application name…); IBTMessageSpaceEx spx= (IBTMessageSpaceEx)sp; Guid spuid= Guid.Empty; Guid agentid= BTXProperties.MessageAgentClassID; agent.RegisterService(ref agentid, null); IBTMessageBatch expex= sp.GetMessageBatch(ref spuid); //new code… IBTMessageAgentFactory msgfactory = (IBTMessageAgentFactory)agent; //creates the message... IBTMessage msg= (IBTMessage)msgfactory.CreateMessage(); msg.Context.Write("MessageType", "http://schemas.microsoft.com/BizTalk/2003/system-properties", "Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage"); msg.Context.Promote("MessageType", "http://schemas.microsoft.com/BizTalk/2003/system-properties", "Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage"); msg.Context.Write("ActivationServiceID", "http://schemas.microsoft.com/BizTalk/2003/system-properties", My Orchestration full name...); msg.Context.Promote("ActivationServiceID", "http://schemas.microsoft.com/BizTalk/2003/system-properties", My Orchestration full name...); msg.Context.Write("ParameterCount", "http://schemas.microsoft.com/BizTalk/2003/xlangs-runtime-properties", 2); msg.OriginatorSID= My user...; msg.OriginatorSID= My PID…; msg.IgnoreRoutingFailures= true; msg.SetPartCount= 3; IBaseMessagePart Param1Part= msgfactory.CreateMessagePart(); IBaseMessagePart Param2Part= msgfactory.CreateMessagePart(); Param1Part.Charset= "utf-8"; Param2Part.Charset= "utf-8"; IBaseMessagePart MtPart= msgfactory.CreateMessagePart(); //fills the meta data... MtPart.Data= GetValueParamTestMetaData(); MtPart.Charset= "utf-8"; MemoryStream Param1Strm= (MemoryStream)GetValueParamTestData("Hello"); MemoryStream Param2Strm= (MemoryStream)GetValueParamTestData("ExecMessage"); Param1Part.Data= Param1Strm; Param2Part.Data= Param2Strm; msg.AddPart("MetaData", MtPart, false); msg.AddPart("Param0", Param1Part, false); msg.AddPart("Param1", Param2Part, false); IBTMessage[] msgs= new IBTMessage[1]{msg}; expex.PostMessages(msgs, 1); //gets a transaction... Type TXType = Type.GetTypeFromCLSID(new Guid("30C1F639-A3A6-4E0A-879D-0B15020754C6")); object TXObj = Activator.CreateInstance(TXType); IXLANGTranDispImpl TXImpl = (IXLANGTranDispImpl) TXObj; ITransaction CurrentTX= (ITransaction)TXImpl.GetTransaction(Guid.NewGuid().ToString(), -1, 1048576); IBTOperationStatus opst= null; try { expex.CommitBatch(CurrentTX, CommitBatchFlags.CB_Default, out opst); txxx.Commit(0, 1, 0); } catch(Exception ex) { Marshal.ReleaseComObject(opst); } |
The black lines represent the adding to our base code. We basically create the MetaData part set the data using the GetValueParamTestMetaData method explained above. We also create two message parts (one for each parameter), sets its stream data and add it to the message. We also set the ParameterCount contextual property and the PartCount message property.
That’s all. With using variants of this code we can pass .Net parameters and execute orchestrations. We can use .Net parameters because in the that we can uses XMessages we have a little dependency with the types generated by BizTalk in the compilation process of our orchestrations.
The types that we use must be serializables since the BizTalk engine have to deserialize those message parts to obtain instances of the parts specified in the MetaData part. It is a typical case that some orchestrations receives generics documents represented by an instance of the System.Xml.XmlDocument class, but this class is not marked as serializable and that’s a problem. To pass XmlDocument parameters to our orchestrations we should use the Microsoft.XLANGs.RuntimeTypes.XmlDocumentSerializationProxy class to wrap the documents. The next code shows how to obtain the MetaData relative to an XmlDocument parameter.
| //constants section… #region consts const string gacassb= "Microsoft.XLANGs.BizTalk.Engine, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35"; const string iFirstPropName= "iFirst"; const string nPartsPropName= "nParts"; const string TypePropName = "Type"; const string ArrayPropName= "Item"; #endregion //this method returns the meta data of an XmlDocument parameter… public Stream GetCustomMetaData() { Assembly btxassb= Assembly.Load(gacassb); Type ParamArrayType= btxassb.GetType("Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamDataArray"); Type ParamDataType= btxassb.GetType("Microsoft.BizTalk.XLANGs.BTXEngine.ExecMessage+ParamData"); object[] args= new object[]{1}; //creates the first param... object ParamMetaData= Activator.CreateInstance(ParamDataType); //set the metadata properties... ParamDataType.GetProperty(iFirstPropName).SetValue(ParamMetaData, 1, null); ParamDataType.GetProperty(nPartsPropName).SetValue(ParamMetaData, 1, null); ParamDataType.GetProperty(TypePropName).SetValue(ParamMetaData, typeof(Microsoft.XLANGs.RuntimeTypes.XmlDocumentSerializationProxy), null); //creates the param list... object ParamsMetaData= Activator.CreateInstance(ParamArrayType, args); ParamArrayType.GetProperty(ArrayPropName).SetValue(ParamsMetaData, ParamMetaData, new object[]{0}); //serializes the meta data... BinaryFormatter formatter= new BinaryFormatter(); MemoryStream buffer= new MemoryStream(); formatter.Serialize(buffer, ParamsMetaData); buffer.Seek(0, SeekOrigin.Begin); return buffer; } |
The required code to uses the XmlDocument-based Meta data is the same that our previous example with the exception that we only have to add one part that contains the Xml documents.
Where are we?
In this article we have shown how to uses the MessageAgent to submit messages to the MessageBox that starts some orchestrations. We also explore different variants to pass the required orchestration parameters. We really think that using the MessageAgent is a nice vehicle to develop powerful low-level BizTalk messaging applications. We hope that with this article we try the subscriptions interiorities much more closely to BizTalk developers.
In future posts we explore how to use the MessageAgent to start activate orchestrations without any interaction with the required adapters.
------------------------------------------------------------------
My friend Didiosky Benítez is an expert in. NET-based technologies and specifically in Web Services and messaging related technologies. He has been working with BizTalk since the 2002 version.