ServiceBusMessageInspector
The next step is to create the ServiceBusMessageInspector. This topic lists the code to create the ServiceBusMessageInspector.
ServiceBusMessageInspector
To bridge the gap between the Service Bus and BizTalk Server, 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. 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. Previously, it is stated that it’s common to have a message with an empty body and the entire payload specified in the user-defined properties. For this reason, it’s important 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 this white paper, 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. This operation cannot be done using a custom pipeline component because a custom pipeline does not provide access to the WCF message. Later, property schemas are created that define 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, add their XML qualified name and value to a collection of key/value pairs. Next, 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 promotes all the properties contained in the collection. Likewise, to write but not promote a set of properties to the BizTalk message context, add their XML qualified name and value to another collection of key/value pairs. Next, 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. So when using a WCF-Custom send port, you are allowed 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 if the message inspector is enabled.
IsTrackingEnabled: Gets or sets a value indicating if 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 should 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 should 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 should be used only when the message inspector is used in a WCF-Custom send port.
You can enable the component tracking and use DebugView to monitor its runtime behavior. For convenience, the following code is for the custom message inspector:
/// <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 = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property";
private const string PropertiesToPromoteKey = "https://schemas.microsoft.com/BizTalk/2006/01/Adapters/WCF-properties/Promote";
private const string PropertiesToWriteKey = "https://schemas.microsoft.com/BizTalk/2006/01/Adapters/WCF-properties/WriteToContext";
private const string ContentTypeProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#ContentType";
private const string CorrelationIdProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#CorrelationId";
private const string LabelProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#Label";
private const string ReplyToProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#ReplyTo";
private const string ReplyToSessionIdProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#ReplyToSessionId";
private const string SessionIdProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/brokered-message-property#SessionId";
private const string ToProperty = "https://schemas.microsoft.com/servicebusforwindowsserver/2013/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>>();
// Retrieve 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)
{
}
#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)
{
}
/// <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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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, StringComparison.OrdinalIgnoreCase) == 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.
var 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
}
The constructor of the message inspector reads the configuration data and initializes three dictionaries that contain 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 promotes or writes only those properties that are specified in the configuration of the WCF receive location using the PropertiesToPromote and PropertiesToWrite properties of the endpoint behavior.
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.
You can enable the component tracking and use DebugView, as shown in the following image, to monitor its runtime behavior:
.jpg "Debug View")
If you stop both orchestrations and use the Service Bus Explorer or client application to send a request message to the requestqueue or requesttopic, the message is read by one of the WCF receive locations and suspended because no subscribers are available. You can use the BizTalk Server Administration Console to examine the context properties; specifically those properties promoted or written by the ServiceBusMessageInspector, highlighted in green (BrokeredMessagePropertySchema) and yellow (PropertySchema) in the following image:
.jpg "Context Properties")
Next Step
See Also
Concepts
ListenUriEndpointBehavior
SessionChannelEndpointBehavior
ServiceBusMessageInspector
TokenProviderEndpointBehavior
Create the Endpoint Behaviors