SalesBuy
1-855-856-7678
Technical SupportSupport
Service Bus Message Inspector
The final step to bridge the gap between the Service Bus and BizTalk Server has been to create a component to translate the properties of a BrokeredMessage into context properties of a BizTalk message and vice versa. Indeed, a WCF-Custom receive location can use the NetMessagingBinding to receive messages from a queue or subscription without the need to use any custom component to extend the normal behavior of the WCF adapter runtime. However, using this approach, a BizTalk application can read only the body of the message retrieved from the Service Bus, not the value of the properties specified by the sender. At the beginning of the article, I highlighted that it’s quite common to have a message with an empty body and the entire payload specified in the user defined properties. For this reason, it’s extremely important, not to say mandatory, to create a component that turns the properties of a BrokeredMessage into context properties of a BizTalk message and vice versa.
In the first part of the article we have seen that a WCF service which uses the NetMessagingBinding to read messages from a queue or a subscription, can retrieve a BrokeredMessageProperty object from the Properties collection of the incoming WCF message. This fact gives us the possibility to write a custom message inspector to intercept the incoming WCF message and read the BrokeredMessageProperty from its properties. For obvious reasons, this operation cannot be done using a custom pipeline component because a custom pipeline does not provide access to the WCF message. In the previous step we created the property schemas which define, respectively, the object specific and user defined properties for Service Bus messages. The last piece to complete the puzzle is to find a way to promote or write a property to the BizTalk message context. Fortunately, the WCF adapter runtime provides an easy way to accomplish this task.
As explained in the BizTalk Server documentation on MSDN, to promote a set of properties to the BizTalk message context you need to add their xml qualified name and value to a collection of key/value pairs. Next you must add a new property to the inbound WCF message whose key must be equal to:
and specify the collection of key/value pairs as its value. The WCF adapter looks for this property in the WCF message and will promote all the properties contained in the collection.
Likewise, to write but not promote a set of properties to the BizTalk message context, you need to add their XML qualified name and value to another collection of key/value pairs. Next you must add another property to the inbound WCF message whose key must be equal to:
and specify the second collection of key/value pairs as its value. The WCF adapter looks for this property in the WCF message and writes all the properties contained in the collection to the BizTalk message context.
The ServiceBusMessageInspector component exploits this feature to promote or write the properties of the BrokeredMessageProperty to the context of the BizTalk message when a WCF-Custom receive location reads a message from a Service Bus queue or subscription. Conversely, the component uses another characteristic of the WCF adapter to retrieve the context property from an outbound WCF message and translate them into properties of a BrokeredMessageProperty. In fact, the send handler of the WCF adapter copies all the message context properties into the Properties collection when it creates a WCF message from a BizTalk message. Therefore, when using a WCF-Custom send port, this allows access to any message context properties within a custom message inspector or channel component at runtime.
-
The ServiceBusMessageInspectorBehaviorExtensionElement can be used to register the message inspector in the machine.config file. It exposes the following properties that you can configure when creating a WCF-Custom receive location or send port:
-
IsComponentEnabled: gets or sets a value indicating whether the message inspector is enabled.
-
IsTrackingEnabled: Gets or sets a value indicating whether the message inspector is enabled.
-
PropertiesToPromote: defines which user defined properties must be extracted from the Properties collection of the BrokeredMessageProperty and promoted to the BizTalk message context. This property is intended to be used only when the message inspector is used in a WCF-Custom receive location
-
PropertiesToWrite: defines which user defined properties must be extracted from the Properties collection of the BrokeredMessageProperty and written but not promoted to the BizTalk message context. This property is intended to be used only when the message inspector is used in a WCF-Custom receive location.
-
PropertiesToSend: defines which context properties must be extracted from the Properties collection of the WCF message and added to the Properties collection of the BrokeredMessageProperty. This property is intended to be used only when the message inspector is used in a WCF-Custom send port.
For you convenience, I included the code of the custom message inspector in the following example.
ServiceBusMessageInspector Class
/// <summary>
/// This class is used to promote\write properties to the message context.
/// </summary>
public class ServiceBusMessageInspector : IDispatchMessageInspector, IClientMessageInspector
{
#region Private Constants
//***************************
// Constants
//***************************
private const string Source = "ServiceBusMessageInspector";
private const string BrokeredMessagePropertySchemaNamespace =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property";
private const string PropertiesToPromoteKey =
"http://schemas.microsoft.com/BizTalk/2006/01/Adapters/WCF-properties/Promote";
private const string PropertiesToWriteKey =
"http://schemas.microsoft.com/BizTalk/2006/01/Adapters/WCF-properties/WriteToContext";
private const string ContentTypeProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#ContentType";
private const string CorrelationIdProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#CorrelationId";
private const string LabelProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#Label";
private const string ReplyToProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#ReplyTo";
private const string ReplyToSessionIdProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#ReplyToSessionId";
private const string SessionIdProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#SessionId";
private const string ToProperty =
"http://schemas.microsoft.com/servicebus/2011/brokered-message-property#To";
private const string ContentType = "ContentType";
private const string CorrelationId = "CorrelationId";
private const string DeliveryCount = "DeliveryCount";
private const string EnqueuedTimeUtc = "EnqueuedTimeUtc";
private const string ExpiresAtUtc = "ExpiresAtUtc";
private const string Label = "Label";
private const string MessageId = "MessageId";
private const string ReplyTo = "ReplyTo";
private const string ReplyToSessionId = "ReplyToSessionId";
private const string ScheduledEnqueueTimeUtc = "ScheduledEnqueueTimeUtc";
private const string SequenceNumber = "SequenceNumber";
private const string SessionId = "SessionId";
private const string TimeToLive = "TimeToLive";
private const string To = "To";
private const string PropertiesToPromote = "PropertiesToPromote";
private const string PropertiesToWrite = "PropertiesToWrite";
private const string PropertiesToSend = "PropertiesToSend";
//***************************
// Messages
//***************************
private const string PropertiesElementCannotBeNull =
"[ServiceBusMessageInspector] The PropertiesElement object cannot be null.";
private const string NameAndNamespaceNumberAreDifferent =
"[ServiceBusMessageInspector] The number of items in the name and namespace comma-separated " +
"lists is different in the {0} property.";
private const string PropertyAddedToList = "[ServiceBusMessageInspector] The following property was added to " +
"the [{0}] list:\n\r - Name=[{1}]\n\r - Namespace=[{2}]\n\r - Value=[{3}]";
private const string ExceptionFormat = "[ServiceBusMessageInspector] The following exception occurred=[{0}].";
#endregion
#region Private Fields
//***************************
// Private Fields
//***************************
private readonly bool isComponentEnabled;
private readonly bool isTrackingEnabled;
private readonly Dictionary<string, string> propertiesToPromoteDictionary = new Dictionary<string, string>();
private readonly Dictionary<string, string> propertiesToWriteDictionary = new Dictionary<string, string>();
private readonly Dictionary<string, string> propertiesToSendDictionary = new Dictionary<string, string>();
#endregion
#region Public Constructors
//***************************
// Public Constructors
//***************************
/// <summary>
/// Initializes a new instance of the ServiceBusMessageInspector class.
/// </summary>
public ServiceBusMessageInspector()
{
isComponentEnabled = true;
}
/// <summary>
/// Initializes a new instance of the ServiceBusMessageInspector class.
/// </summary>
/// <param name="isComponentEnabled">This value indicates whether the message inspector is enabled.</param>
/// <param name="isTrackingEnabled">This value indicates whether tracking is enabled.</param>
/// <param name="propertiesToPromote">This object defines the properties to promote.</param>
/// <param name="propertiesToWrite">This object defines the properties to write.</param>
/// <param name="propertiesToSend">This object defines the properties to send.</param>
public ServiceBusMessageInspector(bool isComponentEnabled,
bool isTrackingEnabled,
PropertiesElement propertiesToPromote,
PropertiesElement propertiesToWrite,
PropertiesElement propertiesToSend)
{
this.isComponentEnabled = isComponentEnabled;
this.isTrackingEnabled = isTrackingEnabled;
// Propulate the dictionary that contains the properties to promote
InitializeDictionary(PropertiesToPromote, propertiesToPromote, propertiesToPromoteDictionary);
// Propulate the dictionary that contains the properties to write
InitializeDictionary(PropertiesToWrite, propertiesToWrite, propertiesToWriteDictionary);
// Propulate the dictionary that contains the properties to send out
InitializeDictionary(PropertiesToSend, propertiesToSend, propertiesToSendDictionary);
}
#endregion
#region IDispatchMessageInspector Members
//**********************************
// IDispatchMessageInspector Members
//**********************************
/// <summary>
/// Called after an inbound message has been received but
/// before the message is dispatched to the intended operation.
/// This method extracts the properties from the BrokeredMessageProperty
/// object contained in the WCF message properties and write or promote
/// them in the message context.
/// </summary>
/// <param name="request">The request message.</param>
/// <param name="channel">The incoming channel.</param>
/// <param name="instanceContext">The current service instance.</param>
/// <returns>The object used to correlate state. This object is passed back
/// in the BeforeSendReply method.</returns>
public object AfterReceiveRequest(ref Message request, IClientChannel channel, InstanceContext instanceContext)
{
try
{
if (!isComponentEnabled)
{
return Guid.NewGuid();
}
var propertiesToPromoteList = new List<KeyValuePair<XmlQualifiedName, object>>();
var propertiesToWriteList = new List<KeyValuePair<XmlQualifiedName, object>>();
// Retreive the BrokeredMessageProperty from the current operation context
var incomingProperties = OperationContext.Current.IncomingMessageProperties;
var brokeredMessageProperty = (BrokeredMessageProperty)incomingProperties[BrokeredMessageProperty.Name];
if (brokeredMessageProperty != null)
{
// Include BrokeredMessageProperty explicit properties
// in the list of properties to write to the BizTalk message context
if (!string.IsNullOrEmpty(brokeredMessageProperty.ContentType))
{
AddItemToList(ContentType,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ContentType,
PropertiesToWrite,
propertiesToWriteList);
}
if (!string.IsNullOrEmpty(brokeredMessageProperty.CorrelationId))
{
AddItemToList(CorrelationId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.CorrelationId,
PropertiesToWrite,
propertiesToWriteList);
}
AddItemToList(DeliveryCount,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.DeliveryCount,
PropertiesToWrite,
propertiesToWriteList);
AddItemToList(EnqueuedTimeUtc,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.EnqueuedTimeUtc,
PropertiesToWrite,
propertiesToWriteList);
AddItemToList(ExpiresAtUtc,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ExpiresAtUtc,
PropertiesToWrite,
propertiesToWriteList);
if (!string.IsNullOrEmpty(brokeredMessageProperty.Label))
{
AddItemToList(Label,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.Label,
PropertiesToWrite,
propertiesToWriteList);
}
if (!string.IsNullOrEmpty(brokeredMessageProperty.MessageId))
{
AddItemToList(MessageId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.MessageId,
PropertiesToWrite,
propertiesToWriteList);
}
if (!string.IsNullOrEmpty(brokeredMessageProperty.ReplyTo))
{
AddItemToList(ReplyTo,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ReplyTo,
PropertiesToWrite,
propertiesToWriteList);
}
if (!string.IsNullOrEmpty(brokeredMessageProperty.ReplyToSessionId))
{
AddItemToList(ReplyToSessionId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ReplyToSessionId,
PropertiesToWrite,
propertiesToWriteList);
}
AddItemToList(ScheduledEnqueueTimeUtc,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ScheduledEnqueueTimeUtc,
PropertiesToWrite,
propertiesToWriteList);
AddItemToList(SequenceNumber,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.SequenceNumber,
PropertiesToWrite,
propertiesToWriteList);
if (!string.IsNullOrEmpty(brokeredMessageProperty.SessionId))
{
AddItemToList(SessionId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.SessionId,
PropertiesToWrite,
propertiesToWriteList);
}
AddItemToList(TimeToLive,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.TimeToLive.TotalSeconds,
PropertiesToWrite,
propertiesToWriteList);
if (!string.IsNullOrEmpty(brokeredMessageProperty.To))
{
AddItemToList(To,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.To,
PropertiesToWrite,
propertiesToWriteList);
}
// Promote or write properties indicated in the
// configuration of the message inspector.
if (brokeredMessageProperty.Properties != null &&
brokeredMessageProperty.Properties.Count > 0)
{
foreach (var property in brokeredMessageProperty.Properties)
{
if (propertiesToPromoteDictionary.ContainsKey(property.Key))
{
AddItemToList(property.Key,
propertiesToPromoteDictionary[property.Key],
property.Value,
PropertiesToPromote,
propertiesToPromoteList);
continue;
}
if (propertiesToWriteDictionary.ContainsKey(property.Key))
{
AddItemToList(property.Key,
propertiesToWriteDictionary[property.Key],
property.Value,
PropertiesToWrite,
propertiesToWriteList);
}
}
}
}
// Notify the WCF Adapter the properties to promote
if (propertiesToPromoteList.Count > 0)
{
request.Properties.Add(PropertiesToPromoteKey, propertiesToPromoteList);
}
// Notify the WCF Adapter the properties to write
if (propertiesToWriteList.Count > 0)
{
request.Properties.Add(PropertiesToWriteKey, propertiesToWriteList);
}
}
catch (Exception ex)
{
Trace.WriteLineIf(isTrackingEnabled, string.Format(ExceptionFormat, ex.Message));
EventLog.WriteEntry(Source, ex.Message, EventLogEntryType.Error);
throw;
}
return Guid.NewGuid();
}
/// <summary>
/// Called after the operation has returned but before the reply message is sent.
/// </summary>
/// <param name="reply">The reply message. This value is null if the operation is one way.</param>
/// <param name="correlationState">The correlation object returned from the AfterReceiveRequest method.</param>
public void BeforeSendReply(ref Message reply, object correlationState)
{
return;
}
#endregion
#region IClientMessageInspector Members
//**********************************
// IClientMessageInspector Members
//**********************************
/// <summary>
/// Enables inspection or modification of a message after a reply message is
/// received but prior to passing it back to the client application.
/// </summary>
/// <param name="reply">The message to be transformed into types and handed
/// back to the client application.</param>
/// <param name="correlationState">Correlation state data.</param>
public void AfterReceiveReply(ref Message reply, object correlationState)
{
return;
}
/// <summary>
/// Enables inspection or modification of a message before a request message is sent to a service.
/// </summary>
/// <param name="message">The message to be sent to the service.</param>
/// <param name="channel">The client object channel.</param>
/// <returns>The object that is returned as the correlationState
/// argument of the AfterReceiveReply method.</returns>
public object BeforeSendRequest(ref Message message, IClientChannel channel)
{
try
{
if (!isComponentEnabled)
{
return message.Headers.Action;
}
// Create a new BrokeredMessageProperty object
var brokeredMessageProperty = new BrokeredMessageProperty();
// Adds properties to the BrokeredMessageProperty object
foreach (var property in message.Properties)
{
if (string.Compare(property.Key, ContentTypeProperty, true) == 0)
{
brokeredMessageProperty.ContentType = message.Properties[ContentTypeProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
ContentType,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ContentType));
continue;
}
if (string.Compare(property.Key, CorrelationIdProperty, true) == 0)
{
brokeredMessageProperty.CorrelationId = message.Properties[CorrelationIdProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
CorrelationId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.CorrelationId));
continue;
}
if (string.Compare(property.Key, LabelProperty, true) == 0)
{
brokeredMessageProperty.Label = message.Properties[LabelProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
Label,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.Label));
continue;
}
if (string.Compare(property.Key, ReplyToProperty, true) == 0)
{
brokeredMessageProperty.ReplyTo = message.Properties[ReplyToProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
ReplyTo,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ReplyTo));
continue;
}
if (string.Compare(property.Key, ReplyToSessionIdProperty, true) == 0)
{
brokeredMessageProperty.ReplyToSessionId = message.Properties[ReplyToSessionIdProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
ReplyToSessionId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.ReplyToSessionId));
continue;
}
if (string.Compare(property.Key, SessionIdProperty, true) == 0)
{
brokeredMessageProperty.SessionId = message.Properties[SessionIdProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
SessionId,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.SessionId));
continue;
}
if (string.Compare(property.Key, ToProperty, true) == 0)
{
brokeredMessageProperty.To = message.Properties[ToProperty] as string;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
To,
BrokeredMessagePropertySchemaNamespace,
brokeredMessageProperty.To));
continue;
}
// Include properties indicated in the
// configuration of the message inspector.
string[] parts = property.Key.Split('#');
if (parts.Length == 2)
{
if (propertiesToSendDictionary.ContainsKey(parts[1]) &&
propertiesToSendDictionary[parts[1]] == parts[0])
{
brokeredMessageProperty.Properties[parts[1]] = property.Value;
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
PropertiesToSend,
parts[0],
parts[1],
property.Value));
}
}
}
// Add the BrokeredMessageProperty to the WCF message properties
message.Properties[BrokeredMessageProperty.Name] = brokeredMessageProperty;
}
catch (Exception ex)
{
Trace.WriteLineIf(isTrackingEnabled, string.Format(ExceptionFormat, ex.Message));
EventLog.WriteEntry(Source, ex.Message, EventLogEntryType.Error);
throw;
}
return Guid.NewGuid();
}
#endregion
#region Private Methods
//******************
// Private Methods
//******************
/// <summary>
/// Initialize the dictionary passed as second parameter
/// with the information read from the first argument.
/// In particular, the Name and Namespace properties of the
/// PropertiesElement object contain respectively
/// the comma-separated list of property names and namespaces
/// of a set of properties. The name of each property will
/// become the key in the dictionary, whereas the namespace
/// will become the corresponding value.
/// </summary>
/// <param name="dictionaryName">The name of the dictionary.</param>
/// <param name="properties">A PropertiesElement object.</param>
/// <param name="dictionary">The dictionary to populate.</param>
private void InitializeDictionary(string dictionaryName,
PropertiesElement properties,
Dictionary<string, string> dictionary)
{
if (properties == null)
{
Trace.WriteLineIf(isTrackingEnabled, PropertiesElementCannotBeNull);
throw new ApplicationException(PropertiesElementCannotBeNull);
}
if (!string.IsNullOrEmpty(properties.Name) &&
!string.IsNullOrEmpty(properties.Namespace))
{
var nameArray = properties.Name.Split(new[] { ',', ';', '|' });
var namespaceArray = properties.Namespace.Split(new[] { ',', ';', '|' });
if (namespaceArray.Length == nameArray.Length)
{
if (nameArray.Length > 0)
{
for (int i = 0; i < nameArray.Length; i++)
{
dictionary[nameArray[i]] = namespaceArray[i];
}
}
}
else
{
Trace.WriteLineIf(isTrackingEnabled, string.Format(NameAndNamespaceNumberAreDifferent,
dictionaryName));
throw new ApplicationException(string.Format(NameAndNamespaceNumberAreDifferent,
dictionaryName));
}
}
}
/// <summary>
/// Adds a new property to the specified list.
/// </summary>
/// <param name="name">The name of a property.</param>
/// <param name="ns">The namespace of a property.</param>
/// <param name="value">The value of a property.</param>
/// <param name="listName">The name of the list.</param>
/// <param name="list">The list to add the property to.</param>
private void AddItemToList(string name,
string ns,
object value,
string listName,
List<KeyValuePair<XmlQualifiedName, object>> list)
{
list.Add(new KeyValuePair<XmlQualifiedName, object>(new XmlQualifiedName(name, ns), value));
Trace.WriteLineIf(isTrackingEnabled, string.Format(PropertyAddedToList,
listName,
name,
ns,
value));
}
#endregion
}
Component Description
The constructor of the message inspector reads the configuration data and initializes three dictionaries that contain, respectively, the properties to promote, to write, and to send out.
-
The AfterReceiveRequest method is executed when a WCF-Custom receive location, configured to use the message inspector, receives a message from the Service Bus. This method performs the following actions:
-
Uses the current OperationContext to read the BrokeredMessageProperty from the Properties collection of the WCF message,
-
Includes the BrokeredMessageProperty properties in the list of the properties to write to the BizTalk message context
-
Reads the user-defined properties from the Properties collection of the BrokeredMessageProperty and promote or write only those that have been specified in the configuration of the WCF receive location using the PropertiesToPromote and PropertiesToWrite properties of the endpoint behavior.
-
Uses the current OperationContext to read the BrokeredMessageProperty from the Properties collection of the WCF message,
The BeforeSendRequest method is executed when a WCF-Custom send port, configured to use the message inspector, is about to send a message to the Service Bus. This method performs the following actions:
-
Creates a new BrokeredMessageProperty object.
-
Reads the context properties, defined in the BrokeredMessagePropertySchema, from the Properties collection of the WCF message and assigns their values to the corresponding properties of the BrokeredMessageProperty.
-
Adds the custom context properties, in our sample defined in the PropertySchema, to the Properties collection of the BrokeredMessageProperty.
-
Finally, adds the BrokeredMessageProperty to the WCF message properties.
Component Tracking
You can enable the component tracking and use DebugView, as shown in the picture below, to monitor its runtime behavior.
If you stop both orchestrations and use the client application to send a request message to BizTalk Server, the BizTalk Server is suspended. This suspension allows you to use the BizTalk Server Administration Console to examine the context properties of the inbound document and, in particular, to review the properties promoted or written to the message context, in order, by the following components:
-
WCF-Custom adapter (see properties at points 2 and 5).
-
ServiceBusMessageInspector (see properties at point 3)
-
Xml Disassembler component within the XMLReceive pipeline (see properties at points 1 and 4).
Build Date:
