Content Based Correlation

This topic applies to Windows Workflow Foundation 4 (WF4).

When workflow services communicate with clients and other services, often there is some data in the exchanged messages that uniquely relates a message to a particular instance. Content-based correlation uses this data in the message, such as a customer number or order ID, to route messages to the proper workflow instance. This topic explains how to use content-based correlation in workflows.

Using Content-Based Correlation

Content-based correlation is used when a workflow service has multiple methods that are accessed by a single client and a piece of data in the exchanged messages identifies the desired instance.

Each messaging activity used in these communications must specify the location of the data in the message that uniquely identifies the instance. This is done by providing a MessageQuerySet, using either a QueryCorrelationInitializer or CorrelatesOn, that queries the message for the piece or pieces of data that uniquely identify the instance.

Caution:

The data that is used to identify the instance is hashed into a correlation key. Care must be taken to ensure that the data used for correlation is unique or else collisions in the hashed key could occur and cause messages to be misrouted. For example, a correlation based solely on a customer name may cause a collision because there may be multiple customers with the same name. The colon (:) should not be used as part of the data used to correlate the message because it is already used to delimit the message query’s key and value to form the string that is subsequently hashed.

In the following example, the initial Receive/SendReply in a workflow service returns an OrderId, which is then passed back by the client on the call to the following Receive activity in the workflow service.

Variable<string> OrderId = new Variable<string>();
Variable<string> Item = new Variable<string>();
Variable<CorrelationHandle> OrderIdHandle = new Variable<CorrelationHandle>();

Receive StartOrder = new Receive
{
    CanCreateInstance = true,
    ServiceContractName = "IOrderService",
    OperationName = "StartOrder"
};

SendReply ReplyToStartOrder = new SendReply
{
    Request = StartOrder,
    Content = SendParametersContent.Create(new Dictionary<string, InArgument>
        { { "OrderId", new InArgument<string>((env) => OrderId.Get(env)) } }),
    CorrelationInitializers =
    {
        new QueryCorrelationInitializer
        {
            CorrelationHandle = OrderIdHandle,
            MessageQuerySet = new MessageQuerySet
            {
                {
                    "OrderId", 
                    new XPathMessageQuery("sm:body()/tempuri:StartOrderResponse/tempuri:OrderId")
                }
            }
        }
    }
};

Receive AddItem = new Receive
{
    ServiceContractName = "IOrderService",
    OperationName = "AddItem",
    CorrelatesWith = OrderIdHandle,
    CorrelatesOn = new MessageQuerySet
    {
        {
            "OrderId", 
              new XPathMessageQuery("sm:body()/tempuri:AddItem/tempuri:OrderId")
        }
    },
    Content = ReceiveParametersContent.Create(new Dictionary<string, OutArgument>
        { { "OrderId", new OutArgument<string>(OrderId) },  
        { "Item", new OutArgument<string>(Item) } })
};

SendReply ReplyToAddItem = new SendReply
{
    Request = AddItem,
    Content = SendParametersContent.Create(new Dictionary<string, InArgument>
        { { "Reply", new InArgument<string>((env) => "Item added: " + Item.Get(env)) } }),
};

// Construct a workflow using StartOrder, ReplyToStartOrder, and AddItem.

The previous example shows a content-based correlation that is initialized by the SendReply. The MessageQuerySet specifies that the data used to identify subsequent messages to this service is the OrderId.

SendReply ReplyToStartOrder = new SendReply
{
    Request = StartOrder,
    Content = SendParametersContent.Create(new Dictionary<string, InArgument>
        { { "OrderId", new InArgument<string>((env) => OrderId.Get(env)) } }),
    CorrelationInitializers =
    {
        new QueryCorrelationInitializer
        {
            CorrelationHandle = OrderIdHandle,
            MessageQuerySet = new MessageQuerySet
            {
                {
                    "OrderId", 
                    new XPathMessageQuery("sm:body()/tempuri:StartOrderResponse/tempuri:OrderId")
                }
            }
        }
    }
};

The Receive activity that follows the SendReply in the workflow follows the correlation that was initialized by the SendReply. Both activities share the same CorrelationHandle, but each one has its own MessageQuerySet and XPathMessageQuery that specifies where the identifying data is in that particular message. On the activity that initializes the correlation, this MessageQuerySet is specified in the CorrelationInitializers property, and for any following Receive activities, it is specified using the CorrelatesOn property.

Receive AddItem = new Receive
{
    ServiceContractName = "IOrderService",
    OperationName = "AddItem",
    CorrelatesWith = OrderIdHandle,
    CorrelatesOn = new MessageQuerySet
    {
        {
            "OrderId", 
              new XPathMessageQuery("sm:body()/tempuri:AddItem/tempuri:OrderId")
        }
    },
    Content = ReceiveParametersContent.Create(new Dictionary<string, OutArgument>
        { { "OrderId", new OutArgument<string>(OrderId) },  
        { "Item", new OutArgument<string>(Item) } })
};

A content-based correlation can be initialized by any messaging activity (Send, Receive, SendReply, ReceiveReply) when the data flows as part of a message. If the particular piece of data does not flow as part of a message, then it can be initialized explicitly by using the InitializeCorrelation activity. If multiple pieces of data are required to uniquely identify the message, then multiple queries can be added to the MessageQuerySet. In these examples, a CorrelationHandle was explicitly provided to each of the activities using the CorrelatesWith or CorrelationHandle properties, but if there is only one correlation required for the entire workflow, such as in this example where everything correlates on OrderId, the implicit correlation handle management provided by WorkflowServiceHost is sufficient.

Using the InitializeCorrelation Activity

In the previous example, the OrderId flowed to the caller through the SendReply activity and this is where the correlation was initialized. The same behavior can be accomplished by using the InitializeCorrelation activity. The InitializeCorrelation activity takes the CorrelationHandle and a dictionary of items that represent the data used to map the message to the correct instance. To use the InitializeCorrelation activity in the preceding sample, remove the CorrelationInitializers from the SendReply activity and initialize the correlation using the InitializeCorrelation activity.

Variable<string> OrderId = new Variable<string>();
Variable<string> Item = new Variable<string>();
Variable<CorrelationHandle> OrderIdHandle = new Variable<CorrelationHandle>();

InitializeCorrelation OrderIdCorrelation = new InitializeCorrelation
{
    Correlation = OrderIdHandle,
    CorrelationData = { { "OrderId", new InArgument<string>(OrderId) } }
};

Receive StartOrder = new Receive
{
    CanCreateInstance = true,
    ServiceContractName = "IOrderService",
    OperationName = "StartOrder"
};

SendReply ReplyToStartOrder = new SendReply
{
    Request = StartOrder,
    Content = SendParametersContent.Create(new Dictionary<string, InArgument> { { "OrderId", new InArgument<string>((env) => OrderId.Get(env)) } }),
};

// Other messaging activities omitted...

The InitializeCorrelation activity is then used in the workflow, after the variables that hold the data are populated but before the Receive activity that correlates with the initialized CorrelationHandle.

// Construct a workflow using OrderIdCorrelation, StartOrder, ReplyToStartOrder,
// and other messaging activities.
Activity wf = new Sequence
{
    Variables =
    {
        OrderId,
        Item,
        OrderIdHandle
    },
    Activities =
    {
        // Wait for a new order.
        StartOrder,
        // Assign a unique identifier to the order.
        new Assign<string>
        {
            To = new OutArgument<string>( (env) => OrderId.Get(env)),
            Value = new InArgument<string>( (env) => Guid.NewGuid().ToString() )
        },
        ReplyToStartOrder,
        // Initialize the correlation.
        OrderIdCorrelation,
        // Wait for an item to be added to the order.
        AddItem,
        ReplyToAddItem
     }
};

Configuring XPath Queries Using the Workflow Designer

In the previous examples, the activities and the XPath queries used in the message queries were specified in code. The workflow designer in Visual Studio 2010 also provides the ability to generate XPaths from DataContract types for content-based correlation. The first XPath configured in the previous example was configured for the SendReply.

SendReply ReplyToStartOrder = new SendReply
{
    Request = StartOrder,
    Content = SendParametersContent.Create(new Dictionary<string, InArgument>
        { { "OrderId", new InArgument<string>((env) => OrderId.Get(env)) } }),
    CorrelationInitializers =
    {
        new QueryCorrelationInitializer
        {
            CorrelationHandle = OrderIdHandle,
            MessageQuerySet = new MessageQuerySet
            {
                {
                    "OrderId", 
                    new XPathMessageQuery("sm:body()/tempuri:StartOrderResponse/tempuri:OrderId")
                }
            }
        }
    }
};

To configure the XPath for a messaging activity in the workflow designer, select the activity in the workflow designer. If the activity is initializing the correlation, as in the previous example, click the ellipsis button for the CorrelationInitializers property in the Properties window. This displays the Add Correlation Initializers dialog window. From this dialog you can specify the correlation type and select the content that is used for the correlation. The CorrelationHandle variable is specified in the Add initializer box, and the correlation type and data used for the correlation is selected from the XPath Queries section of the dialog box.

The second XPath query in the previous example was configured in the Receive activity.

Receive AddItem = new Receive
{
    ServiceContractName = "IOrderService",
    OperationName = "AddItem",
    CorrelatesWith = OrderIdHandle,
    CorrelatesOn = new MessageQuerySet
    {
        {
            "OrderId", 
              new XPathMessageQuery("sm:body()/tempuri:AddItem/tempuri:OrderId")
        }
    },
    Content = ReceiveParametersContent.Create(new Dictionary<string, OutArgument>
        { { "OrderId", new OutArgument<string>(OrderId) },  
        { "Item", new OutArgument<string>(Item) } })
};

To configure the XPath query for a messaging activity that does not initialize the correlation, select the activity in the workflow designer and then click the ellipsis button for the CorrelatesOn property in the Properties window. This displays the CorrelatesOn Definition dialog window.

From this dialog you specify the CorrelationHandle and choose items in the XPath Queries list to build the XPath query.