Windows Server AppFabric Customer Advisory Team

We Have Moved to MSDN

Christian.Martinez — Mon, 07 Feb 2011 21:34:00 GMT

We blog too much! We have maxed out the storage space allowed on MSDN blogs.

All new content will be located on the MSDN Library.

See you there!

Intellisense when editing Service Bus configuration files? Yes please!

Paolo Salvatori — Mon, 07 Feb 2011 19:20:04 GMT

The problem

The setup of Windows Azure AppFabric SDK adds the elements reported in the table below to the configuration contained in the the machine.config files under the following folders:

%windir%\Microsoft.NET\Framework\v4.0.30319\Config
%windir%\Microsoft.NET\Framework64\v4.0.30319\Config

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
 ...
 <system.serviceModel>
  <extensions>
   <behaviorExtensions>
    ...
    <!-- Windows Azure AppFabric Service Bus Endpoint Behaviors -->
    <add name="connectionStatusBehavior" 
         type="Microsoft.ServiceBus.Configuration.ConnectionStatusElement, Microsoft.ServiceBus, ..." />
    <add name="transportClientEndpointBehavior" 
         type="Microsoft.ServiceBus.Configuration.TransportClientEndpointBehaviorElement, ..." />
    <add name="serviceRegistrySettings" 
         type="Microsoft.ServiceBus.Configuration.ServiceRegistrySettingsElement, Microsoft.ServiceBus, ..." />
   </behaviorExtensions>
   <bindingElementExtensions>
    ...
    <!-- Windows Azure AppFabric Service Bus Transport Binding Elements -->
    <add name="tcpRelayTransport" 
         type="Microsoft.ServiceBus.Configuration.TcpRelayTransportElement, Microsoft.ServiceBus, ..." />
    <add name="httpRelayTransport" 
         type="Microsoft.ServiceBus.Configuration.HttpRelayTransportElement, Microsoft.ServiceBus, ..." />
    <add name="httpsRelayTransport" 
         type="Microsoft.ServiceBus.Configuration.HttpsRelayTransportElement, Microsoft.ServiceBus, ..." />
    <add name="onewayRelayTransport" 
         type="Microsoft.ServiceBus.Configuration.RelayedOnewayTransportElement, Microsoft.ServiceBus, ..." />
   </bindingElementExtensions>
   <bindingExtensions>
     ...
    <!-- Windows Azure AppFabric Service Bus Relay Binding -->
    <add name="basicHttpRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.BasicHttpRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
    <add name="webHttpRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.WebHttpRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
    <add name="ws2007HttpRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.WS2007HttpRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
    <add name="netTcpRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.NetTcpRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
    <add name="netOnewayRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.NetOnewayRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
    <add name="netEventRelayBinding" 
         type="Microsoft.ServiceBus.Configuration.NetEventRelayBindingCollectionElement, Microsoft.ServiceBus, ..." />
   </bindingExtensions>
  </extensions>
  ...
 </system.serviceModel>
 ...
</configuration>

However, configuring WCF extensions in the machine.config is not sufficient to enable Intellisense on the relay bindings and endpoint behaviors contained in the AppFabric Service Bus Class Library when you edit the configuration file (web.config, app.config) of an application that exposes or consume services in the cloud. To scope the problem, let’s assume for a moment that you create an on-premises or cloud application that exposes a service via the AppFabric Service Bus. Your service will surely expose one or more endpoints based on one of the relay bindings supplied out-of the-box by the AppFabric Service Bus Class Library. However, when you try to use Visual Studio to define the configuration of any relay bindings and endpoint behaviors provided by the Service Bus, IntelliSense is unable to recognize these elements. For example, the picture below shows what happens when you try to define the configuration for the WebHttpRelayBinding and the TransportClientEndpointBehavior. In a nutshell, you cannot rely on IntelliSense to compose and validate your configuration file using a syntax-driven approach. Moreover, Visual Studio raises a warning for each unrecognized element within your configuration. Now, I don’t know about you, but when I compile my projects I cross my fingers until the build process completes. When I see the Rebuild All succeeded message on the left-bottom corner of Visual Studio I heave a sigh of relief, but in order to attain the maximum level of happiness, my Errors and Warnings lists must be empty!

The solution

Now, the XSD used by Visual Studio IntelliSense is stored under "%ProgramFiles(x86)%\Microsoft Visual Studio 10.0\Xml\Schemas\DotNetConfig.xsd". Hence, I decided to extend this XML schema file to add Intellisense for the bindings and behaviors provided by AppFabric Service Bus. BizTalk Editor came in handy for this purpose because I could use its cut and paste capabilities to accelerate the task. I created an entry for each relay binding or endpoint behavior defined within the AppFabric Service Bus Class Library as you can see in the image below:

After customizing the DotNetConfig.xsd, IntelliSense works as expected as shown in the following picture.

Conclusions

Utilizing IntelliSense rather than cut and paste or memorization makes it more likely I will enter settings correctly and also gets rid of all of those unsightly warnings! Here you can download the Visual Studio 2010 DotNetConfig.xsd customized for the Azure Service Bus. Please drop me a line if you find any problems or improvements to suggest. Thanks!

Review by: Christian Martinez and Tim Wieman.

Enterprise Library on Windows Azure: Key Learnings from Customer Project

Valery M — Mon, 31 Jan 2011 12:55:00 GMT

This post has been moved to a new location. Please follow the link below. Thank you.

http://windowsazurecat.com/2011/02/enterprise-library-on-windows-azure-key-learnings-from-customer-project/

How to use Duplex MEP to communicate with BizTalk from a .NET application or a WF workflow running inside AppFabric Part 3

Paolo Salvatori — Thu, 20 Jan 2011 15:30:00 GMT

Introduction

In the first article of the series we discussed how to exchange messages with an orchestration via a two-way WCF Receive Location using the Duplex Message Exchange Pattern. This form of bi-directional communication is characterized by the ability of both the service and the client to send messages to each other independently either using one-way or request/reply messaging. In a service-oriented architecture or a service bus composed of multiple, heterogeneous systems, interactions between autonomous client and service applications are asynchronous and loosely-coupled. All communications require published and discoverable service contracts, well-known data formats and a shared communication infrastructure. In the second part of the article we saw how to implement an asynchronous communication between a client application and a WCF Workflow Service running within IIS\AppFabric Hosting Services using the Durable Duplex Correlation provided by WF 4.0. Besides, we discussed how to create a custom Activity for extending AppFabric Tracking with user-defined events and how to exploit the XML-based data transformation capabilities provided by the new BizTalk Server Mapper directly in a WF project thanks to the new Mapper Activity contained in the AppFabric Connect. In the final article of the series, we’ll examine how to implement an asynchronous communication between a WCF Workflow Service and an Orchestration using WS-Addressing and Content-Based Correlation.

Before explaining the architecture of the demo, let me briefly introduce and discuss some of the techniques that I used to implement my solution.

Correlation in WF 4.0

If you are a WF or a BizTalk developer, you are surely familiar with the concept of correlation. Typically, at runtime workflows or orchestrations have multiple instances executing simultaneously. Therefore, when a workflow service implements an asynchronous communication pattern to exchange messages with other services, correlation provides the mechanism to ensure that messages are sent to the appropriate workflow instance. Correlation enables relating workflow service messages to each other or to the application instance state, such as a reply to an initial request, or a particular order ID to the persisted state of an order-processing workflow. Workflow Foundation 4.0 provides 2 different categories of correlation called, respectively, Protocol-Based Correlation and Content-Based Correlation. Protocol-based correlations use data provided by the message delivery infrastructure to provide the mapping between messages. Messages that are correlated using protocol-based correlation are related to each other using an object in memory, such as a RequestContext, or by a token provided by the transport protocol. Content-based correlations relate messages to each other using application-specified data. Messages that are correlated using content-based correlation are related to each other by some application-defined data in the message, such as a customer number.

Protocol-Based Correlation

Protocol-based correlation uses the transport mechanism to relate messages to each other and the appropriate workflow instance. Some system-provided protocol correlation mechanisms include Request-Reply correlation and Context-Based correlation. A Request-Reply correlation is used to correlate a single pair of messaging activities to form a two-way synchronous inbound or outbound operation, such as a Send paired with a ReceiveReply, or a Receive paired with a SendReply. The Visual Studio 2010 Workflow Designer also provides a set of activity templates to quickly implement this pattern. A context-based correlation is based on the context exchange mechanism described in the .NET Context Exchange Protocol Specification. To use context-based correlation, a context-based binding such as BasicHttpContextBinding, WSHttpContextBinding or NetTcpContextBinding must be used on the endpoint.

For more information about protocol correlation, see the following topics on MSDN:

For more information about using the Visual Studio 2010 Workflow Designer activity templates, see Messaging Activities. For sample code, see the Durable Duplex and NetContextExchangeCorrelation samples.

Content-Based Correlation

Content-based correlation uses data in the message to associate it to a particular workflow instance. Unlike protocol-based correlation, content-based correlation requires the application developer to explicitly state where this data can be found in each related message. Activities that use content-based correlation specify this message data by using a MessageQuerySet. Content-based correlation is useful when communicating with services that do not use one of the context bindings such as BasicHttpContextBinding. For more information about content-based correlation, see Content Based Correlation. For sample code, see the Content-Based Correlation and Correlated Calculator samples.

A content-based correlation takes data from the incoming message and maps it to an existing instance. This kind of correlation can be used in the following 2 scenarios:

when a WCF 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;
when a WCF workflow service submits a request to a downstream service and asynchronously waits for a response that could potentially arrive after some minutes, hours or days.

In my demo I used 2 different types of protocol-based correlation, respectively, the Durable Duplex Correlation to realize an asynchronous message exchange between the client application and the WCF workflow service and the Content-Based Correlation to implement an asynchronous communication between the WF workflow service and the underlying BizTalk orchestration.

Architecture of the Demo

The following picture depicts the architecture of the demo. The idea behind the application is quite simple: a Windows Forms application submits a question to a WCF workflow service hosted in IIS\AppFabric and asynchronously waits for the related answer. The AsyncMagic8Ball WCF workflow service uses the Mapper activity to transform the incoming request in a format suitable to be consumed by the underlying BizTalk application. For more information on how using the Mapper activity in a WF workflow to implement message transformation, please refer to the second part of this series. Next, the WCF workflow service sends the request message to BizTalk Server via via a one-way WCF-NetTcp Receive Location. In particular, the client endpoint used by the WCF workflow service to transmit the request to BizTalk is configured to use a custom message inspector called ReplyToMessageInspector. At runtime, this component assigns the URL where the WCF workflow service is asynchronously waiting for the response to the ReplyTo header of the outgoing request message. Once received, the request message is read and processed by a new instance of the AsyncMagic8Ball orchestration that returns a response message containing one of 20 standardized answers. In particular, the orchestration copies the request ID from the request to the response message. This piece of information will be used by the WF runtime to correlate the response message back to the appropriate instance of the AsyncMagic8Ball WCF workflow service. Then the orchestration reads the callback URL from the WCF.ReplyToAddress context property and assigns its value to the Address property of the dynamic send port used to return the response to the appropriate instance of the WCF workflow service. Upon receiving the response message from BizTalk, the WCF workflow service applies another map using the Mapper activity and returns the resulting message to the client application.

Message Flow

The Windows Forms Client Application enables a user to specify a question and a delay in seconds. When the user presses the Ask button, a new request message containing the question and the delay is created and sent to a the WCF workflow service. Before sending the first message, the client application creates and opens a service host to expose a callback endpoint that the workflow can invoke to return the response message. In particular, the binding used to expose this callback contract is the NetTcpBinding, whereas the binding used to send the request message to the WCF workflow service is the NetTcpContextBinding. We will expand on this point in the next sections when we’ll analyze the client-side code.
The WCF workflow service receives the request message of type WFRequest using the Receive activity of a ReceiveAndSendReply composed activity. In particular, the Receive activity is used to initialize 2 distinct correlation handles:
- The callbackHandle is configured to use the Durable Duplex Correlation and is used to correlate the incoming request with the response returned to the client application.
- The correlationHandle is instead configured to use the Content-Based Correlation and is used to correlate the outgoing request with the response returned by BizTalk Server.
The WCF workflow service uses the CustomTrackingActivity to keep track of individual processing steps and uses an instance of the Mapper activity to transform the WFRequest object into an instance of the BizTalkRequest class. See the second part of this series for more information on the CustomTrackingActivity and Mapper activities.
The WCF workflow service uses a composed SendAndReceiveReply activity to send the BizTalkRequest message to the WCF-NetTcp receive location exposed by the BizTalk application. The ReplyToMessageInspector assigns the URL where the WCF workflow service is asynchronously waiting for the response to the ReplyTo header of the outgoing request message.
A one-way WCF-NetTcp receive location receives the request message and the XmlReceive pipeline promotes the MessageType context property.
The Message Agent submits the request message to the MessageBox (BizTalkMsgBoxDb).
A new instance of the AsyncMagic8Ball orchestration receives the request message via a one-way logical port and uses a custom helper component called XPathHelper to read the value of the Question and Delay elements from the inbound message.
The AsyncMagic8Ball orchestration invokes the SetResponse static method exposed by the ResponseHelper class to build the response message containing the answer to this question contained in the request message. Then it copies the request ID from the request to the response message, reads the callback URL from the WCF.ReplyToAddress context property and assigns its value to the Address property of the dynamic send port used to return the response to the appropriate instance of the WCF workflow service. The response message is finally published to the MessageBox (BizTalkMsgBoxDb) by the Message Agent.
The response message is retrieved by a one-way Dynamic Send Port.
The PassThruTransmit send pipeline is executed by the Dynamic Send Port.
The response message is returned to the WCF workflow service.
The WCF workflow service uses a Mapper activity to transform the BizTalkResponse object into an instance of the WFRequest class.
The WCF workflow service uses a Send activity to send back the response message to the client application. The Send activity is configured to use the callback correlation that contains the URI of the callback endpoint exposed by the client application.

Client Code

Please refer to the second article of this series for a full explanation of the code used by the client application to invoke the WCF workflow service using the Durable Duplex communication pattern.

The AsyncMagic8Ball WCF Workflow Service

In this section I will focus my attention on how the AsyncMagic8Ball WCF workflow service exchange messages with both the client application and the BizTalk orchestration. When I created the WCF Workflow Service, the initial workflow just contained a Sequence activity with a Receive activity followed by a SendReply activity as shown in the following illustration.

I selected the Sequential activity and clicked the Variables button to display the corresponding editor. Next, I created a variable for each message to exchange with the client and BizTalk application and then I created two CorrelationHandle variables called respectively callbackHandle and correlationHandle. The first of these two variables is used to implement the Durable Duplex Correlation with the client application, whereas the second one is configured to use the Content-Based Correlation and holds the Id of the incoming request. This information is copied into the request sent to BizTalk and is contained in the corresponding response message. Upon receiving a response from BizTalk Server, the WF runtime uses this correlation handle to identify the appropriate instance of the AsyncMagic8Ball WCF workflow service to pass the message to. In order to expose a NetTcpContextBinding endpoint I configured the Receive activity as shown in the following picture:

In particular, I used the ServiceContractName property of the Receive activity to specify the target namespace and the contract interface of the service endpoint and I used the Action property to specify the action header of the request message. To initialize the callback correlation handle, I selected the Receive activity and then I clicked the ellipsis button next to the (Collection) text for the CorrelationInitializers property in the property grid for the Add Correlation Initializers dialog box to appear. As shown in the picture below, I specified callbackHandle as correlation handle and I selected Callback correlation initializer in as correlation initializer.

Before invoking the downstream BizTalk application, the WCF workflow service immediately returns an ACK message to the caller. Therefore, I configured the SendReply activity, bound to the initial Receive activity, to return a WFAck message, as shown in the picture below.

As you can notice, the workflow uses a CustomTrackingActivity to emit a user-defined event. This pattern is used throughout the workflow. Custom tracking records generated at runtime by the WCF workflow service can be analyzed using the AppFabric Dashboard. For more information on the CustomTrackingActivity and how to use the AppFabric Dashboard to monitor the runtime behavior of the WCF workflow service, please read the previous article of this series.

There are two options to invoke the WCF-NetTcp receive location exposed by the BizTalk application:

The first possibility is to generate a custom WCF proxy activity using the Add Service Reference and use this activity to invoke the underlying WCF receive location in a synchronous way. For more information on this technique, please refer to the previous article of this series.
The second alternative is using the the messaging activities provided out-of-the-box by WF and the Content-Based Correlation to implement an asynchronous communication between the WCF workflow service and the underlying orchestration. In this article, we examine this approach.

The following picture depicts the central part of the AsyncMagic8Ball WCF workflow service.

In a nutshell, this section of the workflow executes the following actions:

Uses a SendAndReceiveReply activity to send the request message to the one-way WCF-NetTcp receive location exposed by the BizTalk application.
Tracks a user-defined event using the CustomTrackingActivity.
Uses a Receive activity to receive the response back from BizTalk.

The following figure shows how I configured the Send activity to transmit the request message to BizTalk.

As highlighted above, I used the ServiceContractName property to specify the target namespace and the contract interface utilized by the client endpoint and the EndpointConfigurationName property to define the name of the client endpoint used to transmit the request message to BizTalk. The bizTalkAsyncNetTcpBinding endpoint is defined in the configuration/system.serviceModel/client section of the web.config (the configuration file is shown later in the article). In particular, this endpoint is configured to use a custom message inspector that assigns the callback address of the workflow service to the ReplyTo header of request messages.

Next, I initialized the correlation handle used to implement the content-based correlation. To accomplish this task, I selected the Send activity and then I clicked the ellipsis button next to the (Collection) text for the CorrelationInitializers property in the property grid for the Add Correlation Initializers dialog box to appear. On the left panel of the dialog, I specified correlationHandle as correlation handle and then I selected Query correlation initializer as correlation initializer. Finally, I specified the correlation key by selecting the Id element from the data contract of the BizTalkRequest message.

To complete the configuration of the content-based correlation, I selected the Receive activity and I specified the correlationHandle as value for the CorrelatesWith property. The latter defines the correlation handle that is used to route the message to the appropriate workflow instance, whereas the CorrelatesOn property sets the MessageQuerySet used to query the message to extract correlation data.

To specify the correlation key on the response message, I clicked the ellipsis button next to the (Collection) text for the CorrelatesOn property to open the Add Correlation Initializers dialog box. Then, as shown in the picture below, I selected the Id property from the data contract of the BizTalkResponse message.

The last part of the WCF workflow invokes the callback endpoint exposed by the client application to return the response to the initial request. In particular, the latter contains the Id of the original request, and this allows the client application to correlate the response to the corresponding request, especially when the client has multiple in-flight requests. For more details on how the client handles the callback, please refer to the second part of this series.

This portion of the workflow performs just 2 steps:

Uses a Send activity to send the response message back to the caller. This activity is configured to use the callback handle correlation.
Tracks a user-defined event using the CustomTrackingActivity.

The following figure shows how I configured the Send activity used to transmit the response message back to the caller.

As highlighted above, I assigned to the the callbackHandle, previously initialized, to the CorrelatesWith property. Then I properly set the other properties like OperationName, Action, and ServiceContractName to match the characteristics of the callback service endpoint exposed by the client application.

The ReplyToMessageInspector component

In order to transparently add the ReplyTo header to outgoing request messages, I created a custom message inspector. For your convenience, I include below the code of this component along with the code of endpoint behavior used to register at runtime this extension.

ReplyToBehaviorExtensionElement Class

#region Using Directives
using System;
using System.Configuration;
using System.Diagnostics;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.ServiceModel.Configuration;
#endregion

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.ReplyToHelper
{
    /// <summary>
    /// This ReplyToBehaviorExtensionElement inherits from BehaviorExtensionElement base class
    /// and allows to register the ReplyToEndpointBehavior in the configuration file.
    /// </summary>
    public class ReplyToBehaviorExtensionElement : BehaviorExtensionElement
    {
        #region Private Constants
        private const string AddressProperty = "address";
        private const string EnabledProperty = "enabled";
        private const string TraceEnabledProperty = "traceEnabled";
        #endregion

        #region Protected Methods
        protected override object CreateBehavior()
        {
            return new ReplyToEndpointBehavior(this.Enabled,
                                               this.TraceEnabled,
                                               this.Address);
        }
        #endregion

        #region Public Methods
        public override Type BehaviorType
        {
            get 
            { 
                return typeof(ReplyToEndpointBehavior); 
            }
        }
        #endregion

        #region Configuration Properties
        [ConfigurationProperty(EnabledProperty, DefaultValue = true)]
        public bool Enabled
        {
            get
            {
                return (bool)base[EnabledProperty];
            }
            set
            {
                base[EnabledProperty] = value;
            }
        }

        [ConfigurationProperty(TraceEnabledProperty, DefaultValue = false)]
        public bool TraceEnabled
        {
            get
            {
                return (bool)base[TraceEnabledProperty];
            }
            set
            {
                base[TraceEnabledProperty] = value;
            }
        }

        [ConfigurationProperty(AddressProperty, 
                               DefaultValue = "http://www.w3.org/2005/08/addressing/anonymous")]
        public string Address
        {
            get
            {
                return (string)base[AddressProperty];
            }
            set
            {
                base[AddressProperty] = value;
            }
        }
        #endregion
    }
}

ReplyToEndpointBehavior class

#region Using Directives
using System;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.ServiceModel.Channels;
using System.ServiceModel.Dispatcher;
#endregion

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.ReplyToHelper
{
    /// <summary>
    /// The ReplyToEndpointBehavior class implements the IEndpointBehavior interface 
    /// and adds the ReplyToMessageInspector to the client runtime. 
    /// </summary>
    public class ReplyToEndpointBehavior : IEndpointBehavior
    {
        #region Private Fields
        private bool enabled = true;
        private bool traceEnabled = false;
        private string address = null;
        #endregion

        #region Public Constructors
        public ReplyToEndpointBehavior(bool enabled, 
                                       bool traceEnabled, 
                                       string address)
        {
            this.enabled = enabled;
            this.traceEnabled = traceEnabled;
            this.address = address;
        }
        #endregion

        #region IEndpointBehavior Members
        public void AddBindingParameters(ServiceEndpoint endpoint, BindingParameterCollection bindingParameters)
        {
            return;
        }

        public void ApplyClientBehavior(ServiceEndpoint endpoint, ClientRuntime clientRuntime)
        {
            clientRuntime.MessageInspectors.Add(new ReplyToMessageInspector(enabled, traceEnabled, address));
        }

        public void ApplyDispatchBehavior(ServiceEndpoint endpoint, EndpointDispatcher endpointDispatcher)
        {
            return;
        }
        public void Validate(ServiceEndpoint endpoint)
        {
            return;
        }
        #endregion
    }
}

ReplyToMessageInspector class

#region Using Directives
using System;
using System.Diagnostics;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.ServiceModel.Dispatcher;
#endregion

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.ReplyToHelper
{
    /// <summary>
    /// The ReplyToMessageInspector adds the ReplyTo header to outgoing messages.
    /// </summary>
    public class ReplyToMessageInspector : IDispatchMessageInspector, IClientMessageInspector
    {
        #region Private Constants
        private const string ReplyToFormat = "[ReplyToMessageInspector] ReplyTo header set to {0}.";
        #endregion

        #region Private Fields
        private bool enabled = true;
        private bool traceEnabled = false;
        private string address = null;
        #endregion

        #region Public Constructors
        public ReplyToMessageInspector(bool enabled,
                                       bool traceEnabled,
                                       string address)
        {
            this.enabled = enabled;
            this.traceEnabled = traceEnabled;
            this.address = address;
        }
        #endregion

        #region IDispatchMessageInspector Members
        public object AfterReceiveRequest(ref Message request, 
                                          IClientChannel channel, 
                                          InstanceContext instanceContext)
        {
            throw new NotImplementedException();
        }

        public void BeforeSendReply(ref Message reply, object correlationState)
        {
            throw new NotImplementedException();
        }
        #endregion

        #region IClientMessageInspector Members
        public void AfterReceiveReply(ref Message reply, object correlationState)
        {
            return;
        }

        public object BeforeSendRequest(ref Message request, IClientChannel channel)
        {
            if (enabled &&
                request != null &&
                !string.IsNullOrEmpty(address))
            {
                request.Headers.ReplyTo = new EndpointAddress(address);
                if (traceEnabled)
                {
                    Trace.WriteLine(string.Format(ReplyToFormat, address));
                    Trace.WriteLine(new string('-', 100));
                }
            }
            return request;
        }
        #endregion
    }
}

The following table shows an excerpt from the configuration file used by the AsyncMagic8Ball service and by the SyncMagi8cBall service introduced in the second part of this article. You can find the original configuration file in the companion code for this article. Note in particular the definition of the ReplyToBehaviorExtensionElement component in the configuration/system.serviceModel/extensions section.

<?xml version="1.0" encoding="utf-8"?>
<configuration>
  ...
  <system.serviceModel>
    ...
    <bindings>
      <netTcpBinding>
        <binding name="netTcpBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpBinding>
      <netTcpContextBinding>
        <binding name="netTcpContextBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpContextBinding>
    </bindings>
    <client>
      <endpoint address="net.tcp://localhost:7171/Magic8BallBizTalk/Sync"
                binding="netTcpBinding"
                bindingConfiguration="netTcpBinding"
                contract="Magic8Ball"
                name="bizTalkSyncNetTcpBinding"/>
      <endpoint address="net.tcp://localhost:7172/Magic8BallBizTalk/Async"
                binding="netTcpBinding"
                behaviorConfiguration="replyToBehavior"
                bindingConfiguration="netTcpBinding"
                contract="Magic8Ball"
                name="bizTalkAsyncNetTcpBinding"/>
    </client>
    <services>
      <service name="SyncMagic8Ball">
        <endpoint address="" 
                  binding="basicHttpContextBinding" 
                  contract="IMagic8BallWF" 
                  name="basicHttpBinding_SyncMagic8Ball" />
        <endpoint address="" 
                  binding="netTcpContextBinding" 
                  bindingConfiguration="netTcpContextBinding" 
                  contract="IMagic8BallWF" 
                  name="netTcpBinding_SyncMagic8Ball" />
      </service>
      <service name="AsyncMagic8Ball">
        <endpoint address="" 
                  binding="basicHttpBinding" 
                  contract="IMagic8BallWF" 
                  name="basicHttpBinding_AsyncMagic8Ball" />
        <endpoint address="" 
                  binding="netTcpContextBinding" 
                  bindingConfiguration="netTcpContextBinding" 
                  contract="IMagic8BallWF" 
                  name="netTcpBinding_AsyncMagic8Ball" />
        <endpoint address=""
                  binding="netTcpContextBinding"
                  bindingConfiguration="netTcpContextBinding"
                  contract="IMagic8BallWFCallback"
                  name="netTcpBinding_AsyncMagic8BallCallback" />
      </service>
    </services>
    <behaviors>
      <endpointBehaviors>
        <!-- This behavior configuration is adopted by the client endpoint used by the Send Activity 
             that transmits the request message to the AsyncMagic8Ball orchestration. -->
        <behavior name="replyToBehavior">
          <replyTo address="net.tcp://localhost/Magic8BallWF/AsyncMagic8Ball.xamlx" 
                   enabled="true" 
                   traceEnabled="true" />
        </behavior>
      </endpointBehaviors>
      ...
    </behaviors>
    <serviceHostingEnvironment multipleSiteBindingsEnabled="true" />
    <extensions>
      <behaviorExtensions>
        <!-- It's necessary to register the custom behavior extension element -->
        <add name="replyTo" 
             type="Microsoft.AppFabric.CAT.Samples.DuplexMEP.ReplyToHelper.ReplyToBehaviorExtensionElement, 
                   Microsoft.AppFabric.CAT.Samples.DuplexMEP.ReplyToHelper, 
                   Version=1.0.0.0, Culture=neutral, 
                   PublicKeyToken=80577993de400321" />
      </behaviorExtensions>
    </extensions>
  </system.serviceModel>
  ...
</configuration>

The AsyncMagic8Ball Orchestration

The following picture shows the structure of the AsyncMagic8Ball orchestration.

The orchestration uses a one-way logical port to receive the inbound request message and dynamic send port to return the corresponding response message. The Trace Request Expression Shape contains the following code to extract the information from the request message. The namespace of the LogHelper and XPathHelper static classes have been eliminated for ease of reading.

LogHelper.WriteLine(System.String.Format("[SyncMagic8Ball] Transport: {0}",
                                         RequestMessage(BTS.InboundTransportType)));
id = XPathHelper.GetValue(RequestMessage, 0, "Id Element XPath Expression");
if (!System.Int32.TryParse(XPathHelper.GetValue(RequestMessage, 0, "Delay Element XPath Expression"),
                           out delayInSeconds))
{
    delayInSeconds = 0;
}
LogHelper.WriteLine(System.String.Format("[SyncMagic8Ball] Id: {0}", id));
LogHelper.WriteLine(System.String.Format("[SyncMagic8Ball] Question: {0}",
                                         XPathHelper.GetValue(RequestMessage, 
                                                              0, 
                                                              "Question Element XPath Expression")));
LogHelper.WriteLine(System.String.Format("[SyncMagic8Ball] Delay: {0}", delayInSeconds));

You can use DebugView, as shown in the picture below, to monitor the trace produced by the orchestration and helper components.

Note
My LogHelper class traces messages to the standard output using the capability supplied by the Trace class. This component is primarily intended to be used for debugging a BizTalk application in a test environment, rather than to be used in a production environment. If you are looking for a tracing framework which combines the high performance and flexibility provided by the Event Tracing for Windows (ETW) infrastructure, you can read the following whitepaper by Valery Mizonov:

Instrumentation Best Practices for High Performance BizTalk Solutions.

The value of the Delay Shape is defined as follows:

new System.TimeSpan(0, 0, delayInSeconds);

Therefore, the orchestration waits for the time interval in seconds specified in the request message before returning the response message to the caller. Finally, the table below shows the code used to set the response message. As you can see, the code specifies the value for the context properties exposed by the WCF Adapter. The penultimate line of code reads the callback URL from the WCF.ReplyToAddress context property and assigns its value to the Address property of the dynamic send port used to return the response to the appropriate instance of the AsyncMagic8Ball workflow service, whereas the last line of code specifies to use the WCF-NetTcp to send the response back to the initial caller.

ResponseMessage = null;
Microsoft.AppFabric.CAT.Samples.DuplexMEP.Helpers.ResponseHelper.SetResponse(ResponseMessage, id);
ResponseMessage(WCF.Action) = "AskQuestionResponse";
ResponseMessage(WCF.SecurityMode) = "Transport";
ResponseMessage(WCF.TransportClientCredentialType) = "Windows";
Magic8BallOutPort(Microsoft.XLANGs.BaseTypes.Address) = RequestMessage(WCF.ReplyToAddress);
Magic8BallOutPort(Microsoft.XLANGs.BaseTypes.TransportType) = "WCF-NetTcp";

Testing the Application

To test the application, you can proceed as follows:

Makes sure to start the DuplexMEP BizTalk application.
Open a new instance of the Client Application, as indicated in the picture below.
Enter an existential question like "Why am I here?", "What's the meaning of like?" or "Will the world end in 2012?" in the Question textbox.
Select one of NetTcpEndpointAsyncWF in the Endpoint drop down list.
Specify a Delay in seconds in the corresponding textbox.
Press the Ask button.

Now, if you press the Ask button multiple times in a row, you can easily notice that the client application is called back by the WCF workflow service in an asynchronous way, that in turn invokes the underlying AsyncMagic8Ball orchestration in an asynchronous manner. Therefore, the client application doesn't need to wait for the response to a previous question before posing a new request.

Make some calls and then open the AppFabric Dashboard. This page is composed of three detailed metrics sections: three detailed metrics sections: Persisted WF Instances, WCF Call History, and WF Instance History. These sections display monitoring and tracking metrics for instances of .NET Framework 4 WCF and WF services. Let’s focus our attention on the WF Instance History section, highlighted in red in the figure below. The latter displays historical statistics derived from tracked workflow instance events stored in one or more monitoring databases. It can draw data from several monitoring databases, if the server or farm uses more than one monitoring database for services deployed at the selected scope.

If you click the Completions link you can review the instances of the AsyncMagic8Ball service that completed in the selected period of time. You can use the Query control on the Tracked WF Instances Page to run a simple query and restrict the number of rows displayed in the grid below.

Finally, you can right-click one of the completed WF instances and select View Tracked Events to access the Tracked Events Page where you can examine events generated by WCF and WF services. On this page you can group events by Event Type, as shown in the figure below, and analyze the user-defined events emitted by the current WCF instance using the CustomTrackingActivity that we saw at the beginning of this article.

In particular, you can quickly investigate the details of a selected event in the Details pane, as highlighted in red in the figure above.

Conclusions

In the final article of this 3-posts series we have seen how to implement an asynchronous communication between a WCF Workflow Service and an Orchestration using WS-Addressing and Content-Based Correlation that represent probably the safest and most reliable way to correlate across truly disparate and disconnected applications in an asynchronous manner. As we have observed in the introduction of the present article, the use of asynchronous communication patterns can dramatically improve the scalability and performance of a distributed application platform where multiple systems exchange messages using WCF. In the first part of this series we have examined how to implement an asynchronous message exchange between a .NET application and an orchestration via a two-way WCF receive location using the Duplex Message Exchange Pattern, whereas in the second part we have seen how to realize an asynchronous communication between a client application and a WCF Workflow Service running within IIS\AppFabric Hosting Services using the Durable Duplex Correlation. In the second article we have also seen how to create a custom activity to emit user-defined events, how to use the AppFabric Dashboard to monitor custom tracking events generated by WF services, and finally how to exploit the Mapper activity provided by AppFabric Connect to implement message transformations in a WCF workflow service. I hope this article can provide you with useful ideas on how to improve the scalability and flexibility of your AppFabric/BizTalk applications. Here you can download the companion code for this article. As always, you feedbacks are more than welcome!

Reviewed by Christian Martinez. Thanks mate!

Running .NET4 Windows Workflows in Azure

SurenM — Wed, 12 Jan 2011 16:00:00 GMT

This blog reviews the current (January 2011) set of options available for hosting existing .NET4 Workflow (WF) programs in Windows Azure and also provides a roadmap to the upcoming features that will further enhance support for hosting and monitoring the Workflow programs. The code snippets included below are also available as an attachment for you to download and try it out yourself.

Workflow in Azure – Today

Workflow programs can broadly classified as durable or non-durable (aka non-persisted Workflow Instances). Durable Workflow Services are inherently long running, persist their state, and use correlation for follow-on activities. Non-durable Workflows are stateless, effectively they start and run to completion in a single burst.

Today non-durable Workflows are readily supported by Windows Azure of course with a few configuration/trivial changes. Hosting durable Workflows today is a challenge; since we do not yet have a ‘Windows Server AppFabric’ equivalent for Azure which can persist, manage and monitor the Service. In brief the big buckets of functionality required to host the durable Workflow Services are:

Monitoring store: There is no Event Collection Service available to gather the ETW events and write them to the SQL Azure based Monitoring database. There is also no schema that ships with .NET Framework for creating the monitoring database, and the one that ships with Windows Server AppFabric is incompatible with SQL Azure – an example, the scripts that are provided with Windows Server AppFabric make use of the XML column type which is currently not supported by SQL Azure.
Instance Store: The schemas used by the SqlWorkflowInstanceStore have incompatibilities with SQL Azure. Specifically, the schema scripts require page locks, which are not supported on SQL Azure.
Reliability: While the SqlWorkflowInstanceStore provides a lot of the functionality for managing instance lifetimes, the lack of the AppFabric Workflow Management Service means that you need to manually implement a way to start your WorkflowServiceHosts before any messages are received (such as when you bring up a new role instance or restart a role instance), so that the contained SqlWorkflowInstanceStore can poll for workflow service instances having expired timers and subsequently resume their execution.

The above limitations make it rather difficult to run a durable Workflow Service on Azure – the upcoming release of Azure AppFabric (Composite Application) is expected to make it possible to run durable Workflow Services. In this blog, we will focus on the design approaches to get your non-durable Workflow instances running within Azure roles.

Today you can run your non-durable Workflow on Azure. What this means, is that your Workflow programs really cannot persist their state and wait for subsequent input to resume execution- they must complete following their initial launch. With Azure you can run non-durable Workflows programs in one of the three ways:

Web Role
Worker Roles
Hybrid

The Web Role acts very much like IIS does on premise as an HTTP server, and is easier to configure and requires little code to integrate and is activated by an incoming request. The Worker Role acts like an on-premise Windows Service and is typically used in backend processing scenarios have multiple options to kick off the processing – which in turn add to the complexity. The hybrid approach, which bridges communication between Azure hosted and on-premise resources, has multiple advantages: it enables you to leverage existing deployment models and also enables use of durable Workflows on premise as a solution until the next release. The following sections, succinctly, provide details on these three approaches and in the ‘Conclusion’ section we will also provide you pointers on the appropriateness of each approach.

Host Workflow Services in a Web Role

The Web Role is similar to a ‘Web Application’ and can also provide a Service perspective to anything that uses the http protocol - such as a WCF service using basicHttpBinding. The Web Role is generally driven by a user interface – the user interacts with a Web Page, but a call to a hosted Service can also cause some processing to happen. Below are the steps that enable you to host a Workflow Service in a Web Role.

First step is to create a Cloud Project in Visual Studio, and add a WCF Service Web Role to it. Delete the IService1.cs, Service1.svc and Service1.svc.cs added by the template since they are not needed and will be replaced by the workflow service XAMLX.

To the Web Role project, add a WCF Workflow Service. The structure of your solution is now complete (see the screenshot below for an example), but you need to add a few configuration elements to enable it to run on Azure.

Windows Azure does not include a section handler in its machine.config for system.xaml.hosting as you have in an on-premises solution. Therefore, the first configuration change (HTTP Handler for XAMLX and XAMLX Activation) is to add the following to the top of your web.config, within the configuration element:

<configSections>

  <sectionGroup name="system.xaml.hosting" type="System.Xaml.Hosting.Configuration.XamlHostingSectionGroup, System.Xaml.Hosting, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35">

    <section name="httpHandlers" type="System.Xaml.Hosting.Configuration.XamlHostingSection, System.Xaml.Hosting, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35" />

  </sectionGroup>

</configSections>

Next, you need to add XAML http handlers for WorkflowService and Activities root element types by adding the following within the configuration element of your web.config, below the configSection that we included above:

<system.xaml.hosting>

  <httpHandlers>

    <add xamlRootElementType="System.ServiceModel.Activities.WorkflowService, System.ServiceModel.Activities, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35" httpHandlerType="System.ServiceModel.Activities.Activation.ServiceModelActivitiesActivationHandlerAsync, System.ServiceModel.Activation, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35" />

    <add xamlRootElementType="System.Activities.Activity, System.Activities, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35" httpHandlerType="System.ServiceModel.Activities.Activation.ServiceModelActivitiesActivationHandlerAsync, System.ServiceModel.Activation, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35" />

  </httpHandlers>

</system.xaml.hosting>

Finally, configure the WorkflowServiceHostFactory to handle activation for your service by adding a serviceActivation element to system.serviceModel\serviceHostingEnvironment element:

<serviceHostingEnvironment multipleSiteBindingsEnabled="true" >

  <serviceActivations>

    <add relativeAddress="~/Service1.xamlx" service="Service1.xamlx"  factory="System.ServiceModel.Activities.Activation.WorkflowServiceHostFactory"/>       

  </serviceActivations>

</serviceHostingEnvironment>

The last step is to deploy your Cloud Project and with that you now have your Workflow service hosted on Azure – graphic below!

Note: Sam Vanhoutte from CODit in his blog also elaborates on Hosting workflow services in Windows Azure and focuses on troubleshooting configuration by disabling custom errors-- do review.

Host Workflows in a Worker Role

The Worker Role is similar to a Windows Service and would start up ‘automatically’ and be running all the time. While the Workflow Programs could be initiated by a timer, it could use other means to activate such as a simple while (true) loop and a sleep statement. When it ‘ticks’ it performs work. This is generally the option for background or computational processing.

In this scenario you use Workflows to define Worker Role logic. Worker Roles are created by deriving from the RoleEntryPoint Class and overriding a few of its method. The method that defines the actual logic performed by a Worker Role is the Run Method. Therefore, to get your workflows executing within a Worker Role, use WorkflowApplication* or WorkflowInvoker to host an instance of your non-service Workflow (e.g., it doesn’t use Receive activities) within this Method. In either case, you only exit the Run Method when you want the Worker Role to stop executing Workflows.

The general strategy to accomplish this is to start with an Azure Project and add a Worker Role to it. To this project you Add a reference to an assembly containing your XAML Workflow types. Within the Run Method of WorkerRole.cs, you initialize one of the host types (WorkflowApplication or WorkflowInvoker), referring to an Activity type contained in the referenced assembly. Alternatively, you can initialize one of the host types by loading an Activity instance from the XAML Workflow file available on the file system.. You will also need to add references to .NET Framework assemblies (System.Activities and System.Xaml - if you wish to load XAML workflows from a file).

Host Workflow (Non-Service) in a Worker Role

For ‘non-Service’ Workflows, your Run method needs to describe a loop that examines some input data and passes it to a workflow instance for processing. The following shows how to accomplish this when the Workflow type is acquired from a referenced assembly:

public override void Run()

{

    Trace.WriteLine("WFWorker entry point called", "Information");

    while (true)

    {

        Thread.Sleep(1000);

        /* ...

         * ...Poll for data to hand to WF instance...

         * ...

         */

        //Create a dictionary to hold input data

        Dictionary<;string, object> inputData = new Dictionary<string,object>();

        //Instantiate a workflow instance from a type defined in a referenced assembly

        System.Activities.Activity workflow = new Workflow1();

        //Execute the WF passing in parameter data and capture output results

        IDictionary<;string, object> outputData =  

        System.Activities.WorkflowInvoker.Invoke(workflow, inputData);

        Trace.WriteLine("Working", "Information");

    }

}

Alternatively, you could perform the above using the WorkflowApplication to host the Workflow instance. In this case, the main difference is that you need to use semaphores to control the flow of execution because the workflow instances will be run on threads separate from the one executing the Run method.

public override void Run()

{

    Trace.WriteLine("WFWorker entry point called", "Information");

    while (true)

    {

        Thread.Sleep(1000);

        /* ...

            * ...Poll for data to hand to WF...

            * ...

            */

        AutoResetEvent syncEvent = new AutoResetEvent(false);

        //Create a dictionary to hold input data and declare another for output data

        Dictionary<;string, object> inputData = new Dictionary<string,object>();

        IDictionary<;string, object> outputData;

        //Instantiate a workflow instance from a type defined in a referenced assembly

        System.Activities.Activity workflow = new Workflow1();

        //Run the workflow instance using WorkflowApplication as the host.

        System.Activities.WorkflowApplication workflowHost = new System.Activities.WorkflowApplication(workflow, inputData);

        workflowHost.Completed = (e) =>

            {

                outputData = e.Outputs;

                syncEvent.Set();

            };

        workflowHost.Run();

        syncEvent.WaitOne();

        Trace.WriteLine("Working", "Information");

    }

}

Finally, if instead of loading Workflow types from a referenced assembly, you want to load the XAML from a file available, for example, one included with WorkerRole or stored on an Azure Drive, you would simply replace the line that instantiates the Workflow in the above two examples with the following, passing in the appropriate path to the XAML file to XamlServices.Load:

System.Activities.Activity workflow = (System.Activities.Activity) 

            System.Xaml.XamlServices.Load(@"X:\workflows\Workflow1.xaml");

By and large, if you are simply hosting logic described in a non-durable workflow, WorkflowInvoker is the way to go. As it offers fewer lifecycle features (when compared to WorkflowApplication), it is also more light weight and may help you scale better when you need to run many workflows simultaneously.

Host Workflow Service in a Worker Role

When you need to host Workflow Service in a Worker Role, there are a few more steps to take. Mainly, these exist to address the fact that Worker Role instances run behind a load balancer. From a high-level, to host a Workflow Service means creating an instance of a WorkflowServiceHost based upon an instance of an Activity or WorkflowService defined either in a separate assembly or as a XAML file. The WorkflowService instance is created and opened in the Worker Role’s OnStart Method, and closed in the OnStop Method. It is important to note that you should always create the WorkflowServiceHost instance within the OnStart Method (as opposed to within Run as was shown for non-service Workflow hosts). This ensures that if a startup error occurs, the Worker Role instance will be restarted by Azure automatically. This also means the opening of the WorkflowServiceHost will be attempted again.

Begin by defining a global variable to hold a reference to the WorkflowServiceHost (so that you can access the instance within both the OnStart and OnStop Methods).

public class WorkerRole : RoleEntryPoint

{

    System.ServiceModel.Activities.WorkflowServiceHost wfServiceHostA;

    …

}

Next, within the OnStart Method, add code to initialize and open the WorkflowServiceHost, within a try block. For example:

public override bool OnStart()

{

    Trace.WriteLine("Worker Role OnStart Called.");

    //…

    try

    {

        OpenWorkflowServiceHostWithAddressFilterMode();

    }

    catch (Exception ex)

    {

        Trace.TraceError(ex.Message);

        throw;

    }

    //…

    return base.OnStart();

}

Let’s take a look at the OpenWorkflowServiceHostWithAddressFilterMode method implementation, which really does the work. Starting from the top, notice how either an Activity or WorkflowService instance can be used by the WorkflowServiceHost constructor, they can even be loaded from a XAMLX file on the file-system. Then we acquire the internal instance endpoint and use it to define both the logical and physical address for adding an application service endpoint using a NetTcpBinding. When calling AddServiceEndpoint on a WorkflowServiceHost, you can specify either just the service name as a string or the namespace plus name as an XName (these values come from the Receive activity’s ServiceContractName property).

private void OpenWorkflowServiceHostWithAddressFilterMode()

{

    //workflow service hosting with AddressFilterMode approach

    //Loading from a XAMLX on the file system

    System.ServiceModel.Activities.WorkflowService wfs =

        (System.ServiceModel.Activities.WorkflowService)System.Xaml.XamlServices.Load("WorkflowService1.xamlx");

    //As an alternative you can load from an Activity type in a referenced assembly:

    //System.Activities.Activity wfs = new WorkflowService1();

    wfServiceHostA = new System.ServiceModel.Activities.WorkflowServiceHost(wfs);

    IPEndPoint ip =         RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["WorkflowServiceTcp"].IPEndpoint;

    wfServiceHostA.AddServiceEndpoint(System.Xml.Linq.XName.Get("IService", "http://tempuri.org/"),

        new NetTcpBinding(SecurityMode.None),

        String.Format("net.tcp://{0}/MyWfServiceA", ip));

    //You can also refer to the implemented contract without the namespace, just passing the name as a string:

    //wfServiceHostA.AddServiceEndpoint("IService",

    //    new NetTcpBinding(SecurityMode.None),

    //    String.Format("net.tcp://{0}/MyWfServiceA", ip));

    wfServiceHostA.ApplyServiceMetadataBehavior(String.Format("net.tcp://{0}/MyWfServiceA/mex", ip));

    wfServiceHostA.ApplyServiceBehaviorAttribute();        

    wfServiceHostA.Open();

    Trace.WriteLine(String.Format("Opened wfServiceHostA"));

}

In order to enable our service to be callable externally, we next need to add an Input Endpoint that Azure will expose at the load balancer for remote clients to use. This is done within the Worker Role configuration, on the Endpoints tab. The figure below shows how we have defined a single TCP Input Endpoint on port 5555 named WorkflowServiceTcp. It is this Input Endpoint, or IPEndpoint as it appears in code, that we use in the call to AddServiceEndpoint in the previous code snippet. At runtime, the variable ip provides the local instance physical address and port which the service must use, to which the load balancer will forward messages. The port number assigned at runtime (e.g., 20000) is almost always different from the port you specify in the Endpoints tab (e.g., 5555), and the address (e.g., 10.26.58.148) is not the address of your application in Azure (e.g., myapp.cloudapp.net), but rather the particular Worker Role instance.

It is very important to know that currently, Azure Worker Roles do not support using HTTP or HTTPS endpoints (primarily due to permissions issues that only Worker Roles face when trying to open one). Therefore, when exposing your service or metadata to external clients, your only option is to use TCP.

Returning to the implementation, before opening the service we add a few behaviors. The key concept to understand is that any workflow service hosted by an Azure Worker Role will run behind a load balancer, and this affects how requests must be addressed. This results in two challenges which the code above solves:

How to properly expose service metadata and produce metadata which includes the load balancer’s address (and not the internal address of the service hosted within a Worker Role instance).
How to configure the service to accept messages it receives from the load balancer, that are addressed to the load balancer.

To reduce repetitive work, we defined a helper class that contains extension methods for ApplyServiceBehaviorAttribute and ApplyServiceMetadataBehavior that apply the appropriate configuration to the WorkflowServiceHost and alleviate the aforementioned challenges.

//Defines extensions methods for ServiceHostBase (useable by ServiceHost &; WorkflowServiceHost)

public static class ServiceHostingHelper

{

    public static void ApplyServiceBehaviorAttribute(this ServiceHostBase host)

    {

        ServiceBehaviorAttribute sba = host.Description.Behaviors.Find<;ServiceBehaviorAttribute>();

        if (sba == null)

        {

            //For WorkflowServices, this behavior is not added by default (unlike for traditional WCF services).

            host.Description.Behaviors.Add(new ServiceBehaviorAttribute() { AddressFilterMode = AddressFilterMode.Any });

            Trace.WriteLine(String.Format("Added address filter mode ANY."));

        }

        else

        {

            sba.AddressFilterMode = System.ServiceModel.AddressFilterMode.Any;

            Trace.WriteLine(String.Format("Configured address filter mode to ANY."));

        }

    }

    public static void ApplyServiceMetadataBehavior(this ServiceHostBase host, string metadataUri)

    {

        //Must add this to expose metadata externally

        UseRequestHeadersForMetadataAddressBehavior addressBehaviorFix = new UseRequestHeadersForMetadataAddressBehavior();

        host.Description.Behaviors.Add(addressBehaviorFix);

        Trace.WriteLine(String.Format("Added Address Behavior Fix"));

        //Add TCP metadata endpoint. NOTE, as for application endpoints, HTTP endpoints are not supported in Worker Roles.

        ServiceMetadataBehavior smb = host.Description.Behaviors.Find<;ServiceMetadataBehavior>();

        if (smb == null)

        {

            smb = new ServiceMetadataBehavior();

            host.Description.Behaviors.Add(smb);

            Trace.WriteLine("Added ServiceMetaDataBehavior.");

        }

        host.AddServiceEndpoint(

            ServiceMetadataBehavior.MexContractName,

            MetadataExchangeBindings.CreateMexTcpBinding(),

            metadataUri

        );

    }

}

Looking at how we enable service metadata in the ApplyServiceMetadataBehavior method, notice there are three key steps. First, we add the UseRequestHeadersForMetadataAddressBehavior. Without this behavior, you could only get metadata by communicating directly to the Worker Role instance, which is not possible for external clients (they must always communicate through the load balancer). Moreover, the WSDL returned in the metadata request would include the internal address of the service, which is not helpful to external clients either. By adding this behavior, the WSDL includes the address of the load balancer. Next, we add the ServiceMetadataBehavior and then add a service endpoint at which the metadata can be requested. Observe that when we call ApplyServiceMetadataBehavior, we specify a URI which is the service’s internal address with mex appended. The load balancer will now correctly route metadata requests to this metadata endpoint.

The rationale behind the ApplyServiceBehaviorAttribute method is similar to ApplyServiceMetadataBehavior. When we add a service endpoint by specifying only the address parameter (as we did above), the logical and physical address of the service are configured to be the same. This causes a problem when operating behind a load balancer, as messages coming from external clients via the load balancer will be addressed to the logical address of the load balancer, and when the instance receives such a message it will not accept—throwing an AddressFilterMismatch exception. This happens because the address in the message does not match the logical address at which the endpoint was configured. With traditional code-based WCF services, we could resolve this simply by decorating the service implementation class with [ServiceBehavior(AddressFilterMode=AddressFilterMode.Any)], which allows the incoming message to have any address and port. This is not possible with Workflow Services (as there is no code to decorate with an attribute), hence we have to add it in the hosting code.

If allowing an incoming address concerns you, an alternative to using AddressFilterMode is simply to specify the logical address that is to be allowed. Instead of adding the ServiceBehaviorAttribute, you simply open the service endpoint specifying both the logical (namely the port the load balancer will receives messages on) and physical address (at which your service listens). The only complication, is that your Workflow Role instance does not know which port the load balancer is listening- so you need to add this value to configuration and read it from their before adding the service endpoint. To add this to configuration, return to the Worker Role’s properties, Settings tab. Add string setting with the value of the of the port you specified on the Endpoints tab, as we show here for the WorkflowServiceEndpointListenerPort.

With that setting in place, the rest of the implementation is fairly straightforward:

private void OpenWorkflowServiceHostWithoutAddressFilterMode()

{

    //workflow service hosting without AddressFilterMode

    //Loading from a XAMLX on the file system

    System.ServiceModel.Activities.WorkflowService wfs =        (System.ServiceModel.Activities.WorkflowService)System.Xaml.XamlServices.Load("WorkflowService1.xamlx");

    wfServiceHostB = new System.ServiceModel.Activities.WorkflowServiceHost(wfs);

    //Pull the expected load balancer port from configuration...

    int externalPort = int.Parse(RoleEnvironment.GetConfigurationSettingValue("WorkflowServiceEndpointListenerPort"));

    IPEndPoint ip = RoleEnvironment.CurrentRoleInstance.InstanceEndpoints["WorkflowServiceTcp"].IPEndpoint;

    //Use the external load balancer port in the logical address...

    wfServiceHostB.AddServiceEndpoint(System.Xml.Linq.XName.Get("IService", "http://tempuri.org/"),

        new NetTcpBinding(SecurityMode.None),

        String.Format("net.tcp://{0}:{1}/MyWfServiceB", ip.Address, externalPort),

        new Uri(String.Format("net.tcp://{0}/MyWfServiceB", ip)));

    wfServiceHostB.ApplyServiceMetadataBehavior(String.Format("net.tcp://{0}/MyWfServiceB/mex", ip));

    wfServiceHostB.Open();

    Trace.WriteLine(String.Format("Opened wfServiceHostB"));

}

With that, we can return to the RoleEntryPoint definition of our Worker Role and override the Run and OnStop Methods. For Run, because the WorkflowServiceHost takes care of all the processing, we just need to have a loop that keeps Run from exiting.

public override void Run()

{

    Trace.WriteLine("Run - WFWorker entry point called", "Information");

    while (true)

    {

        Thread.Sleep(30000);

    }

}

For OnStop we simply close the WorkflowServiceHost.

public override void OnStop()

{

    Trace.WriteLine(String.Format("OnStop - Called"));

    if (wfServiceHostA != null)

        wfServiceHostA.Close();

    base.OnStop();

}

With OnStart, Run and OnStop Methods defined, our Worker Role is fully capable of hosting a Workflow Service.

Hybrid Approach - Host Workflow On-Premise and Reach From the Cloud

Unlike ‘pure’ cloud solutions, hybrid solutions have a set of “on-premises” components: business processes, data stores, and services. These must be on-premises, possibly due to compliance or deployment restrictions. A hybrid solution is one which has parts of the solution deployed in the cloud while some applications remain deployed on-premises.

This is a great interim approach, leveraging on-premise Workflows hosted within on-premise Windows Server AppFabric (as illustrated in the diagram below) to various components and application that are hosted in Azure. This approach may also be applied if stateful/durable Workflows are required to satisfy scenarios. You can build a Hybrid solution and run the Workflows on-premise and use either the AppFabric Service Bus or Windows Azure Connect to reach into your on-premise Windows Server AppFabric instance.

Source: MSDN Blog Hybrid Cloud Solutions with Windows Azure AppFabric Middleware

Conclusion

How do you choose which approach to take? The decision ultimately boils down to your specific requirements, but here are some pointers that can help.

Hosting Workflow Services in a Web Role is very easy and robust. If your Workflow is using Receive Activities as part of its definition, you should be hosting in a Web Role. While you can build and host a Workflow Service within a Worker Role, you take on the responsibility of rebuilding the entire hosting infrastructure provided by IIS in the Web Role- which is a fair amount of non-value added work. That said, you will have to host in a Worker Role when you want to use a TCP endpoint, and a Web Role when you want to use an HTTP or HTTPS endpoint.

Hosting non-service Workflows that poll for their tasks is most easily accomplished within a Worker Role. While you can build another mechanism to poll and then call Workflow Services hosted in a Web Role, the Worker Role is designed to support and keep a polling application alive. Moreover, if your Workflow design does not already define a Service, then you should host it in a Worker Role-- as Web Role hosting would require you to modify the Workflow definition to add the appropriate Receive Activities.

Finally, if you have existing investments in Windows Server AppFabric as hosted Services that need to be called from Azure hosted applications, then taking a hybrid approach is a very viable option. One clear benefit, is that you retain the ability to monitor your system’s status through the IIS Dashboard. Of course this approach has to be weighed against the obvious trade-offs of added complexity and bandwidth costs.

The upcoming release of Azure AppFabric Composite Applications will enable hosting Workflow Services directly in Azure while providing feature parity to Windows Server AppFabric. Stay tuned for the exciting news and updates on this front.

Sample

The sample project attached provides a solution that shows how to host non-durable Workflows, in both Service and non-Service forms. For non-Service Workflows, it shows how to host using a WorkflowInvoker or WorkflowApplication within a Worker Role. For Services, it shows how to host both traditional WCF service alongside Workflow services, in both Web and Worker Roles.

StreamInsight: Creating a LINQPad data context

Mark Simms [MSFT] — Wed, 05 Jan 2011 18:09:00 GMT

This blog post is for those folks comfortable with creating a data stream from an IEnumerable, IObservable or StreamInsight adapter data source. If you’d like more details on creating those, have a look at this blog post, which covers creating an IEnumerable ‘canned’ data source then wrapping a LINQPad context around it.

Creating a data context is very straightforward:

Create a new C# class library project in Visual Studio 2010 called SampleContext.
- Ensure that the Framework type is set to .NET Framework 4 (and NOT .NET Framework 4 Client Profile)
- Add a reference to Microsoft.ComplexEventProcessing.dll
- Add a reference to Microsoft.ComplexEventProcessing.Observable.dll
  - This is required for the IObservable<T>.ToPointStream extension method.
- Add a reference to Microsoft.ComplexEventProcessing.Adapters.dll
  - Optional – this is only required if you have an adapter embedded inside of your context
- Add a reference to System.Reactive.dll
  - Optional – this is only required for creating different types of IObservable streams.
  - Note: System.Reactive.dll is part of the Reactive Extensions framework, which needs to be downloaded separately.
Derive a class from StreamInsightContext in the StreamInsightLinqPad.Samples assembly (available as part of the LINQPad driver for StreamInsight download).
Expose your stream as a property of the class.

Download the sample project here.

That’s it! Here’s what this looks like in practice. First we establish the basic class

 using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using StreamInsightLinqPad.Samples;
    using Microsoft.ComplexEventProcessing;
    using Microsoft.ComplexEventProcessing.Linq;
    using StreamInsight.Samples.Adapters.DataGenerator;
 
    /// <summary>
    /// Define a sample LINQPad context for StreamInsight
    /// </summary>
    public class SampleContext : StreamInsightContext
    {
        public SampleContext(Server server)
            : base(server)
        { }
    }
 

Now we’ll go ahead and expose three types of streams. The first will be one generated from an IEnumerable.

 SimpleEvent[] events = new SimpleEvent[]
{
    new SimpleEvent { Timestamp = new DateTime(2011, 1, 5, 12, 0, 0), ID = 5, Message = "Test" },
    new SimpleEvent { Timestamp = new DateTime(2011, 1, 5, 13, 0, 0), ID = 6, Message = "Test2" },
    new SimpleEvent { Timestamp = new DateTime(2011, 1, 5, 14, 0, 0), ID = 7, Message = "Test3" },
    new SimpleEvent { Timestamp = new DateTime(2011, 1, 5, 15, 0, 0), ID = 5, Message = "Test4" },
};
    
/// <summary>
/// Expose a stream created from an IEnumerable
/// </summary>
public CepStream<SimpleEvent> SimpleStream
{
    get
    {
        return events.ToPointStream(this.Application, t =>
            PointEvent.CreateInsert<SimpleEvent>(t.Timestamp, t),
            AdvanceTimeSettings.IncreasingStartTime);
    }
}
 

Next, we’ll use IObservable (created via Observer.Interval) to generate a “live” stream of events:

 /// <summary>
/// Expose a stream created from an IObservable
/// </summary>
public CepStream<SimpleEvent> ObservableStream
{
    get
    {
        var rand = new Random();
 
        // Create a simple observable that returns a random event every
        // 250 ms
        var interval = Observable.Interval(TimeSpan.FromMilliseconds(250))
            .Select(i => new SimpleEvent
            {
                ID = rand.Next(10),
                Timestamp = DateTime.UtcNow,
                Message = "Observable message!"
            });
        return interval.ToPointStream(Application, s =>
            PointEvent.CreateInsert(s.Timestamp, s),
            AdvanceTimeSettings.IncreasingStartTime,
            null);
    }
}
 

Finally, we’ll leverage the DataGenerator sample adapter (from http://streaminsight.codeplex.com/) to demonstrate creating a LINQPad stream from an adapter.

 /// <summary>
/// Expose a stream created from an adapter instance
/// </summary>
public CepStream<GeneratedEvent> AdapterStream
{
    get
    {
        var generatorConfig = new GeneratorConfig()
        {
            CtiFrequency = 1,
            DeviceCount = 3,
            EventInterval = 250,
            EventIntervalVariance = 10,
            MaxValue = 100
        };
 
        var inputStream = CepStream<GeneratedEvent>.Create(Application,
            "inputStream", typeof(GeneratorFactory), generatorConfig,
            EventShape.Point);
        return inputStream;
    }
}
 

To test, we’ll need to configure LINQPad with our assembly as a data context. To do this:

Add Connection, then select Microsoft StreamInsight from the list of drivers.
Select Custom Context from the Context Kind dropdown.
Click on the ellipsis for Assembly Path (...), then browse to the directory containing SampleContext.dll.
Click OK to finish selecting the new context.

The context streams should now be visible in LINQPad, similar to the screenshot below. Create a new Query, and select C# Statements as the Language, and StreamInsight: SampleContext as the database.

Let’s dump our streams, using this LINQ code:

 // Dump out the enumerable stream.  Since this is an IEnumerable, it 
// will eventually complete and return
SimpleStream.ToEnumerable().Dump("Enumerable stream");
 
// Dump out the observable stream.  As this is an IObservable without
// an end condition, we will have to stop the query (as it will not
// stop on its own)
ObservableStream.ToObservable().Dump("Observable stream");
 

This will show us our enumerable and observable streams:

Finally, let’s dump out some data from our adapter stream:

 // Dump out the adapter stream.  As this is an adapter without
// an end condition, we will have to stop the query (as it will not
// stop on its own)
AdapterStream.ToObservable().Dump("Adapter stream");
 

Resulting in:

Note – I had to make one change to the adapter code to make it work with ToObservable. The GeneratedEvent definition doesn’t define a default constructor (which is needed by ToObservable). In GeneratedEvent.cs, on line 23 I added the code:

 public GeneratedEvent()
{ }
 

That’s it – a LINQPad data context in a nutshell.

StreamInsight: Creating a custom data context for LINQPad, chock full of weathery goodness.

Mark Simms [MSFT] — Wed, 05 Jan 2011 02:14:00 GMT

Following on the heels of my last query pattern blog post, I started to dig into creating a custom data context for use with StreamInsight and LINQPad. The default contexts supplied out of box with the StreamInsight driver (including the Hitchhiker’s Guide context) are great, but don’t contain a large volume of data. I wanted to put together a much larger context to have a baseline to play with larger and more complicated query patterns. My goals for the context/data streams were:

Contain “real” process data, around a million “real” data records with accompanying metadata.
Be easily transferrable / downloadable
Be (relatively) human readable
Be performant (sustain at least 60k events read / sec on my desktop)

Building on the fun I had putting together the windchill sample, here’s what I decided on:

Use weather data from Environment Canada, who has kindly made a great deal of information available for download in XML format.
- This includes both the “live” weather and reference data about the producing weather stations. I chose to use airport weather information, as it is a lot more granular than your average weather station.
Create a custom data context which can pull information from compressed CSV files, and expose it as a CepStream.

If you’re already comfortable with exposing data via IEnumerable interface, check out this blog post that provides a concise summary of creating data contexts.

Why did I choose to store the canned data set as a compressed CSV file? Based on a series of performance explorations, this turned out to be the best choice between file size (being downloadable), and performance (throughput).

For all of the gory performance exploration, check out this blog post.

This blog post walks through the various steps of creating the custom data context:

Preparing a canned data set (Downloading, parsing and formatting).
Reading back the canned data set
Creating a LINQPad data context which exposes the data set as a CepStream<T>.

If you’re comfortable with creating/accessible canned data sets, and you just want to create a custom LINQPad context for use with StreamInsight, skip directly to that section.

Before you start reading through the blog, download and follow along with the code samples:

Complete package (weather context and LINQ samples)

Preparing a canned data set – Downloading

Environment Canada rocks. They really do (caveat: I am Canadian ). They’ve made a great deal of climatic data available for download in their National Climate Archives On-line. From the National Climate Archives site, I chose to download ten year’s worth of information for ten airports in Canada (airport weather stations tend to record and archive data in 30 minute or hourly increments – most “other” weather stations seem to record 2-3x per day). Since the goal is lots of time-series data to look at weather stations it is (hourly records gives us 8760 records per year per weather station. 10 stations times 10 years gives us 876k records.. close enough ).

You can see the available stations in a given area and timeframe by looking at their customized search. Being familiar with Canadian airport geography, I selected them by:

Searching by province for a specific date.
Selecting the data for an airport (I happened to know what they were).
From the specific screen, there’s a navigation option to download data in either CSV or XML format. I chose to download in XML (this was before I had settled on using CSV as the store format of choice).
- The station code and station metadata is also available from the URL for this view (for example, the station code for St. John’s, Newfoundland is 6720.

The web URL for downloading data was of the form:

 var urlQuery = "http://www.climate.weatheroffice.gc.ca/climateData/bulkdata_e.html? timeframe=1&Prov=XX&StationID={0}&Year={1}&Month={2}&Day={3}&format={4}&type=hly";
 

With the following variables:

( 0 ) Station ID
( 1 ) Year
( 2 ) Month
( 3 ) Day
( 4 ) Format (XML, CSV)

Next step was to script up a downloader that pulled down data for 2010 for each of my target weather stations. Threw something quick together in LINQPad to accomplish this. Each invocation of this URL will return data for the given month (i.e. the day value is ignored).

Download the LINQ script Download_WeatherData.

After running this script, I had available to me roughly 1.0 GB of raw XML weather data. The contents took the form:

 <stationinformation>
  <name>OTTAWA MACDONALD-CARTIER INT&apos;L  A</name>
  <province>ONTARIO</province>
  <latitude>45.32</latitude>
  <longitude>-75.67</longitude>
  <elevation>114.00</elevation>
  <climate_identifier>6106000</climate_identifier>
  <wmo_identifier>71628</wmo_identifier>
  <tc_identifier>YOW</tc_identifier>
  <note>All times are specified in Local Standard Time (LST). Add 1 hour to adjust for Daylight Saving Time where and when it is observed.</note>
</stationinformation>
<stationdata day="1" hour="0" minute="0" month="1" year="2010" quality=" ">
  <temp description="Temperature" units="°C">-5.30</temp>
  <dptemp description="Dew Point Temperature" units="°C">-6.00</dptemp>
  <visibility description="Visibility" units="km">4.80</visibility>
  <relhum description="Relative Humidity" units="%">95.00</relhum>
  <winddir description="Wind Direction" units="10&apos;s deg">9.00</winddir>
  <windspd description="Wind Speed" units="km/h">15.00</windspd>
  <stnpress description="Station Pressure" units="kPa">100.03</stnpress>
  <humidex description="Humidex"></humidex>
  <windchill description="Wind Chill">-11.00</windchill>
  <weather description="Weather">Snow</weather>
</stationdata>
 

Note that we have both the station metadata and the “live” weather data available to us. The next step will be to render this data into two separate data sets:

Sorted list of weather data across all of the weather stations, saved as a CSV file with headers.
List of weather station information (one record for each station), saved as a CSV file with headers.

Note that we will also have to shift the timestamp on each data record to the appropriate DateTimeOffset. We’ll use another handy little LINQ nugget to do this (I’ve really gotten into using LINQPad as my general purpose C# scratchpad).

Download the LINQ script Parse_WeatherData.

Let’s walk through the main method before diving into the two key functions, ParseWeatherData and CsvSequenceStore.WriteEvents<T>. We grab the list of files from the download directory (in this case we just take 5 to run quickly and work with a smaller amount of data for test purposes). Next, iterate over the list of files, parsing each and inserting the results into the weatherData and stations lists. The weatherData list is then sorted before being written into the output file via the CsvSequenceStore. Note how we use the GZipStream to transparently write out a compressed stream.

Note that the entire 1 GB worth of data is read into memory before being written out to the destination file. This isn’t the world’s best coding practice. What I should have done was:

Written out each weather data file to a common output .CSV file
Sorted the .CSV file.
Compressed the .CSV file.

However, as this was a one-shot, and I have lots of memory on my desktop, I brute-forced it.

 void Main()
{            
    // Get a list of all of the downloaded XML files.
    var fileList = Directory.GetFiles(@"C:\temp\weatherdata\")
        .Take(5) /* Take the first 5 for testing purposes */
        .Where(d => d.EndsWith(".xml"));
 
    // Define the output file
    var outputFile = @"C:\temp\weather_data_out.csv.gz";
    var stationFile = @"C:\temp\station_data_out.csv.gz";
 
    // Create holding lists for weather data and stations        
    var weatherData = new List<WeatherReading>();
    var stations = new List<StationData>();
 
    // Iterate through the file list and parse for weather data
    var fileCount = (double)fileList.ToList().Count();            
    var progress = 0.0;
    fileList.ToList().ForEach(fl =>
    {                                
        Console.WriteLine("{0:00.0} Parsing file {1}",
            (progress / fileCount * 100.0), fl);
        ParseWeatherData(fl, weatherData, stations);
        progress++;
    });
 
    // Sort the weather data into strict ascending time order
    weatherData = weatherData.OrderBy(d => d.Timestamp).ToList();
 
    // Use the CSV store
    CsvSequenceStore store = new CsvSequenceStore();
 
    // Write the data to a compressed file using the GZipStream class.  Way cool.
    Console.WriteLine("Writing {0} events to file {1}",
        weatherData.Count, outputFile);
 
    using (FileStream fs = new FileStream(outputFile, FileMode.Create))
    {
        using (GZipStream zipStream = new GZipStream(
          fs, CompressionMode.Compress))
        {
            // Write out the weather data set to the file, using the Timestamp field
            // as the event timestamp
            store.WriteEvents<WeatherReading>(zipStream, weatherData,
                e => e.Timestamp);
        }
    }
 
    using (FileStream fs = new FileStream(stationFile, FileMode.Create))
    {
        using (GZipStream zipStream = new GZipStream(
          fs, CompressionMode.Compress))
        {
            // Write out the station data set to the file, using a hard coded
            // timestamp
            store.WriteEvents<StationData>(zipStream, stations,
                e => DateTime.Parse("2010-01-01 00:00:00"));
        }
    }
}
 

Now to dive into the first of the two key functions being called – ParseWeatherData.

  
/// Parse through the file, extracting weather and station data      
void ParseWeatherData(string fileName,
    List<WeatherReading> weatherData, List<StationData> stations)
{
    if (!System.IO.File.Exists(fileName))
        throw new ArgumentException("Could not load file" + fileName);
 
    using (StreamReader sr = new StreamReader(fileName))
    {
        var xml = XElement.Load(sr);
 
        // Parse the station information
        var q = from e in xml.Elements() where e.Name == "stationinformation" select e;
        var stationInfo = q.First();
        var stationName = GetValue<string>(stationInfo, "name");
        var stationCode = GetValue<int>(stationInfo, "wmo_identifier");
 
        var newStation = new StationData()
        {
            Name = stationName,
            Elevation = GetValue<double>(stationInfo, "elevation"),
            Latitude = GetValue<double>(stationInfo, "latitude"),
            Longitude = GetValue<double>(stationInfo, "longitude"),
            TCID = GetValue<string>(stationInfo, "tc_identifier"),
            StationCode = GetValue<int>(stationInfo, "wmo_identifier"),
            Province = GetValue<string>(stationInfo, "province")
        };
 
        // If this is a new station, add it to the list
        if (!stations.Where(s => s.Name == stationName).Any())
            stations.Add(newStation);
 
        // Parse the station data
        var sds = from e in xml.Elements() where e.Name == "stationdata" select e;
        sds.ToList().ForEach((e) =>
        {
            if (!String.IsNullOrEmpty(GetValue<string>(e, "temp")))
            {
                // Assign the core values
                var wr = new WeatherReading()
                {
                    StationCode = stationCode,
                    Temperature = GetValue<double>(e, "temp"),
                    DewpointTemperature = GetValue<double>(e, "dptemp"),
                    Humidex = GetValue<double>(e, "humidex"),
                    RelativeHumidity = GetValue<double>(e, "relhum"),
                    StationPressure = GetValue<double>(e, "stnpress"),
                    Visibility = GetValue<double>(e, "visibility"),
                    WindChill = GetValue<double>(e, "windchill"),
                    WindDirection = GetValue<double>(e, "winddir"),
                    WindSpeed = GetValue<double>(e, "windspd"),
                };
 
                // Convert the timestamp into universal time based
                // on the current station information (lookup based on
                // province                        
                var tzi = tzMap[newStation.Province];
                
                // Asemble the timestamp
                 var dt = new DateTimeOffset(
                    GetAttribute<int>(e, "year"),
                    GetAttribute<int>(e, "month"),
                    GetAttribute<int>(e, "day"),
                    GetAttribute<int>(e, "hour"),
                    GetAttribute<int>(e, "minute"),
                    0, tzi.BaseUtcOffset);
                wr.Timestamp = dt.UtcDateTime;
                                          
                // Parse weather conditions - WeatherConditions is an enum
                // flags field
                string[] conditions = GetValue<string>(e, "weather")
                    .Replace(' ', '_')
                    .Split(new char[] { ',' });
 
                WeatherConditions cond = WeatherConditions.None;
                WeatherConditions total = WeatherConditions.None;
                foreach (var s in conditions)
                {
                    if (Enum.TryParse<WeatherConditions>(s, out cond))
                    {
                        total |= cond;
                    }
                }
 
                // Add the data to the list                    
                weatherData.Add(wr);
            }
        });
    }
}
 

This looks like a lot, but it’s simply an XML parser, leveraging the XElement class. After opening up the file, we pull the file into an XElement object, then pluck out the station and weather (station data) information. The GetValue<T> and GetAttribute<T> functions are simple helpers to ease extracting typed data from the various nodes and coercing the types.

Note that I use Convert.ChangeType in the GetValue<T> and GetAttribute<T> functions. This is a pretty slow method – I would have been better off using a switch() statement on the data type and XmlConvert for performance reasons.

However, as this is code that only runs once, I didn’t bother to add the performance optimization. I did spend a lot of time on performance optimizations for reading the data, which we’ll cover in the next section.

There are only two “custom” operations in ParseWeatherData, to handle two custom data parsing condtions:

Timezone offset. Each of the timestamps in the file is in local time. In order to adjust to UTC time we need to determine the time zone (based on the province field in the stations information, which we use to look up a TimeZoneInfo object from a static dictionary I defined at the bottom of the linq file). This is used as an argument to DateTimeOffset to shift the date to UTC.
Weather conditions. The input XML file can define weather conditions as having multiple values. To support this in StreamInsight in a relatively straightforward fashion, I defined a bitwise [Flags] enum (WeatherConditions) that I fill in on lines 72-83.

Finally, we have our friendly neighborhood public void WriteEvents<T>(Stream stream, IEnumerable<T> events, Func<T, DateTimeOffset> dtFunc) method, which writes out each event to a .CSV file. It takes as arguments:

Stream stream. The destination stream – taking a stream rather than a file name made it easy to substitute the GZipStream writer to easily add in compression.
IEnumerable<T> events. The list of events to write out to the file.
Func<T,DatetimeOffset> dtFunc. In order to easily stream events into StreamInsight, we need to know the temporal properties of the event (i.e. the StartTime). Since the CsvSequenceStore is generic, we hand in a function that identifies the temporal properties (in our case, returning e.Timestamp).

 public void WriteEvents<T>(Stream stream, IEnumerable<T> events, 
        Func<T, DateTimeOffset> dtFunc)
    {
        var encodingArray = new byte[16000];
        var fields = typeof(T).GetProperties();
        var lastField = fields.Last();
 
        var sb = new StringBuilder();
        var str = String.Empty;
        var byteLen = 0;
 
        foreach (var e in events)
        {
            sb.Clear(); str = String.Empty; byteLen = 0;
 
            // Write out the event kind, start time and end time                
            sb.Append("Point,");
            sb.Append(dtFunc(e).Ticks);
            sb.Append(",,");
 
            foreach (var o in fields)
            {
                sb.Append(o.GetValue(e, null).ToString());
                if (o != lastField)
                    sb.Append(",");
            }
            sb.AppendLine();
            str = sb.ToString();
 
            byteLen = Encoding.ASCII.GetBytes(str, 0, str.Length, encodingArray, 0);
            stream.Write(encodingArray, 0, byteLen);
        }
    }
 

This is a very straightforward function. Using reflection, we obtain the list of public properties for the type (via GetProperties()), and use a StringBuilder to assemble the text. Wanted to avoid using String.Join(), hence remembering lastField. Then iterate through the list of properties, and use the reflection method GetValue to write out the string representation to the line.

As a base Stream class only understands writing out bytes, we convert the line into a byte array and write it out.

Note that I ASCII encoding here. Really should be using Unicode encoding, but I happen to know that the data is pure ASCII (at least it is THIS time). Should a future blog feature Unicode data, this line will change

After executing the script, I have a compressed .CSV file containing weather data. Let’s have a look at the contents. First step will be to unzip it into a plain .csv file (via the gzip.exe utility – I could write a script to do this, but we’ll get to using GzipStream to uncompress in the next section).

Now, we crack open our .CSV files with note and have a look (not that these only have 5 weather files in them, as I left the .Take() method uncommented on line 5).

Point,633979188000000000,,71628,1/1/2010 5:00:00 AM,-5.3,-6,95,9,15,4.8,100.03,0,-11,0

Excellent – we now have (mostly) human-readable data in a downloadable (i.e. small) format that we can extract quickly. On to extraction!

Reading back the canned data set

We’ll stay in LINQPad for the time being (will switch over to Visual Studio when it comes time to create the actual data context – feel free to work through this code in VS.NET if it’s more comfortable). Now that we have the canned data set (as our compressed .CSV file), we need to be able to read it back out again in a form that can be turned into a CepStream<T>.

Download the LINQ script Read_WeatherData.

This is very similar, yet a little simpler, than the code which we used in the last section to write the data into the compressed CSV file. The parsing code is the same (also using the CsvSequenceStore class, only now using the ReadEvent<T> method in place of the WriteEvents<T> method. The new piece that has been introduced is the assign function, which defines how we map from the CSV format into POCO (plain old CLR object) format.

This doesn’t seem like a terribly generic way of implementing this, and its not. The reason I’m using a function delegate, not reflection (via PropertyInfo.SetValue() and Convert.ChangeType()) is that we’re now in performance critical code and those two methods were killing performance.

I’ll write the code for dynamically generating this function in a later blog, but for now if you’re interested in the performance numbers, head on over to my performance blog on this topic.

void Main()
{
    CsvSequenceStore store = new CsvSequenceStore();
     
    long count = 0;
 
    Action<string[], WeatherReading> assign = (str, e) =>
    {
        e.StationCode = Int32.Parse(str[0]);
        e.Timestamp = DateTime.Parse(str[1]);
        e.Temperature = Double.Parse(str[2]);
        e.DewpointTemperature = Double.Parse(str[3]);
        e.RelativeHumidity = Double.Parse(str[4]);
 
        e.WindDirection = Double.Parse(str[5]);
        e.WindSpeed = Double.Parse(str[6]);
        e.Visibility = Double.Parse(str[7]);
        e.Humidex = Double.Parse(str[8]);
        e.WindChill = Double.Parse(str[9]);
        e.WeatherConditions = Int32.Parse(str[9]);
    };
        
    using (FileStream fs = new FileStream(@"C:\temp\weather_data_out.csv.gz", FileMode.Open))
    {
        using (GZipStream zipStream = new GZipStream(
            fs, CompressionMode.Decompress))
        {
            var evts2 = store.ReadEvents<WeatherReading>(zipStream, assign);
            var weatherStream = evts2.Select (e => e.Payload)
                    .Take(50);
            weatherStream.Dump("weather events");                
        }
    }
}

Executing this simple main function gives us the shiny weather events (50 of them, as limited by the Take() method on line 30):

The core of this approach lies in the ReadEvents<T> method, which is remarkably simple given the amount of code we’ve written to get to this point . Remembering that each line of CSV consists of the event shape (Point, Interval), the Start and End times, then the payload. We use a wrapper object, TypedEventPayload<T> to encapsulate the payload along with the event characteristics.

public IEnumerable<TypedEventPayload<T>> ReadEvents<T>(Stream stream,
        Action<string[], T> assign) where T : new()
    {
        StreamReader sr = new StreamReader(stream);
        String line = null;
        var fields = typeof(T).GetProperties();
        var lastField = fields.Last();
 
        while ((line = sr.ReadLine()) != null)
        {
            if (line.StartsWith("EventShape,"))
                continue;
            string[] tokens = line.Split(new char[] { ',' });
            TypedEventPayload<T> ep = new TypedEventPayload<T>();
            ep.Payload = new T();
 
            ep.Shape = (EventShape)Enum.Parse(typeof(EventShape), tokens[0]);
            ep.StartTime = new DateTimeOffset(Int64.Parse(tokens[1]), TimeSpan.FromSeconds(0));
            if (String.IsNullOrEmpty(tokens[2]))
                ep.EndTime = ep.StartTime.AddTicks(1);
            else
                ep.EndTime = new DateTimeOffset(Int64.Parse(tokens[2]), TimeSpan.FromSeconds(0));
            
            assign(tokens.Skip(3).ToArray(), ep.Payload);        
 
            yield return ep;
        }
    }

We now have an IEnumerable<T> which we can convert into a CepStream<T> via use of the ToPointStream method.

The really critical piece here is returning the events as an IEnumerable<T>, not as an array or list. If we had to load the entire file into memory before sending a single event to StreamInsight, that wouldn’t be performant from either a memory or a latency perspective. Instead, as we pull and deserialize events from the CSV file we yield them back.

This will allow the CepStream<T> to consume events as they are raised (which we’ll do in the next step).

Finally, we convert the IEnumerable<T> into a CepStream<T> and run some basic queries over it:

var evts2 = store.ReadEvents<WeatherReading>(zipStream, assign);
    var weatherStream = evts2.Select (e => e.Payload)                                        
        .ToPointStream(Application, t =>
            PointEvent.CreateInsert(t.Timestamp, t),
            AdvanceTimeSettings.IncreasingStartTime);
        
    var query = from e in weatherStream
        .TumblingWindow(TimeSpan.FromDays(1000), HoppingWindowOutputPolicy.ClipToWindowEnd)
    select new
    {
        Count = e.Count()
    };

Excellent – we now have a CepStream<T> – the only remaining step to having a canned data set is simple packaging

Wait! You’re probably looking at the IEnumerable<UntypedEventPayload> ReadEvents method and thinking to yourself “Mark.. that’s dumb. That won’t work, because of the IEnumerable state machine, and how Dispose is handled!”. And you’d be right.

Here’s the problem:

The method ReadEvents assumes that someone else is responsible for initializing and disposing of the stream (and any underlying resources – such as the file handle).
IEnumerable, and the corresponding yield statement, create an internal state machine when compiled (i.e. the enumeration uses deferred execution – the behavior we want). The IEnumerable is not finished with the underlying stream until it has enumerated through all of the available values (in our case read to the end of the file).
As such, the lifetime of the Stream needs to be tied to the lifetime of the enumerator (as only the code that manages the enumerator knows when it is complete).
Since this design has the stream managed separately from the enumerator, the underlying stream could be closed before the enumerator even begins to evaluate.

Our current code base works, as we force the entire query to complete before exiting the using block that creates the stream. In a LINQPad context wherein we need to raise the stream from a function and have no direct control over the lifetime of a query this will have the effect of closing the underyling file before any events are read.

We need to redesign our code to encapsulate the file reading into the same class that exposes IEnumerable, and handles stream cleanup correctly.

Reading back the canned data set – the Right Way!

Now, to take the prototype we built out in the last section, and put it in a form suitable for embedding into a library (which the LINQPad context will leverage). We’ll create a base class, SequenceBase, which encapsulates creating streams from files (including compressed files). This will allow us to more easily plug in different types of “serializers”.

SequenceBase is very straightforward, simply providing a helper function to create a new stream upon demand (to allow the enumerator to wrap it in a using() statement).

public abstract class SequenceBase 
{        
    private string _fileName;
    private bool _compress;
    private bool _open;
 
    protected Stream GetStream()
    {
        // If opening the file,     Open/read data/allow others to read
        // If writing a new file,   Create/write data/deny others access
        FileMode fMode = (_open) ? FileMode.Open : FileMode.Create;
        FileAccess aMode = (_open) ? FileAccess.Read : FileAccess.Write;
        FileShare fShare = (_open) ? FileShare.Read : FileShare.None;
 
        // Set the compression mode
        CompressionMode cMode = (_open) ? CompressionMode.Decompress
            : CompressionMode.Compress;
 
        // Create the base file stream            
        var fs = new FileStream(_fileName, fMode, aMode, fShare);
        Stream _stream = fs;
 
        // If using Gzip compression, wrap the file stream
        if (_compress)
        {
            var zs = new System.IO.Compression.GZipStream(fs, cMode);
            _stream = zs;
        }
        return _stream;
    }
 
    public SequenceBase(string fileName, bool compress, bool open)
    {
        _fileName = fileName;
        _compress = compress;
        _open = open;           
    }    
}

With this as a foundation, our Csv reader class becomes very straightforward to implement – essentially taking the contents of the ReadEvents<T> method from the previous section, and wrapping them up in a class derived from SequenceBase.

public class CsvSequenceReaderTyped<T> :
    SequenceBase, IEnumerable<IntervalEvent<T>>
    where T : new()
{
    protected Action<string[], T> _assign;
 
    public CsvSequenceReaderTyped(string fileName, bool compress,
        Action<string[], T> assign) : base(fileName, compress, true)
    {
        _assign = assign;
    }
 
    public IEnumerator<IntervalEvent<T>> GetEnumerator()
    {
        using (var stream = GetStream())
        {
            using (var sr = new StreamReader(stream))
            { 
                String line = null;
                var fields = typeof(T).GetProperties();
                var lastField = fields.Last();
 
                while ((line = sr.ReadLine()) != null)
                {
                    if (line.StartsWith("EventShape,"))
                        continue;
                    string[] tokens = line.Split(new char[] { ',' });
                                                
                    var shape = (EventShape)Enum.Parse(typeof(EventShape), tokens[0]);
                    var startTime = new DateTimeOffset(Int64.Parse(tokens[1]), TimeSpan.FromSeconds(0));
                    DateTimeOffset endTime;
 
                    if (String.IsNullOrEmpty(tokens[2]))
                        endTime = startTime.AddTicks(1);
                    else
                        endTime = new DateTimeOffset(Int64.Parse(tokens[2]), TimeSpan.FromSeconds(0));
 
                    var ep = IntervalEvent<T>.CreateInsert(
                        startTime, endTime, new T());
                        
                   ��_assign(tokens.Skip(3).ToArray(), ep.Payload);
 
                    yield return ep;
                }
            }
        }
    }
 
    System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }
}

Note that we assume that the assignment Action<string[], T> is handed in when the reader is created.

The really critical piece here is creating (via GetStream()) and disposing of (via the using statement) of our underlying file resources each time (which required that the resource management code – opening and closing files – be merged with the resource using code).

That’s it – now that we have our IEnumerable<T> wrapped up in a class that can close out the underlying file stream on Dispose(), we are ready to wrap it up in a LINQPad data context.

Creating a LINQPad data context which exposes the data set as a CepStream<>

Creating the custom data context will be the shortest section of this blog post . With a data source (either an IEnumerable<T>, IObservable<T> or a StreamInsight adapter) ready to go, all that stands between you and LINQPad data context goodness is exposing the CepStream<T> object in a class.

Really – that’s it. Here are the steps:

Create a class that derives from StreamInsight Context.
- StreamInsightContext is defined in StreamInsightLinqPad.Samples.dll, which is typically found in C:\ProgramData\LINQPad\Drivers\DataContext\4.0\StreamInsightLinqPad (3d3a4b0768c9178e)
- Expose the two streams (weather data and station data), based on the code we explored in the reading back the canned data section).

Or, to look at the code:

public class WeatherContext : StreamInsightContext
{
    private static readonly string weatherFile;
    private static readonly string stationFile;
 
    static WeatherContext()
    {
        // Look for the data files in the same directory as this assembly
        var assemblyDir = Path.GetDirectoryName(
            Assembly.GetExecutingAssembly().Location);
        weatherFile = Path.Combine(assemblyDir, @"weather_data_out.csv.gz");
        stationFile = Path.Combine(assemblyDir, @"station_data_out.csv.gz");
    }
 
    public WeatherContext(Server server)
        : base(server)
    { }
 
    public CepStream<WeatherReading> SmallWeather
    {
        get
        {
            return new CsvSequenceReaderTyped<WeatherReading>
                (weatherFile, true, WeatherReading.GetFill()).Take(50)
                    .ToStream(Application, AdvanceTimeSettings.IncreasingStartTime);
        }
    }
 
    public CepStream<WeatherReading> LargeWeather
    {
        get
        {
            return new CsvSequenceReaderTyped<WeatherReading>
                (weatherFile, true, WeatherReading.GetFill())
                .ToStream(Application, AdvanceTimeSettings.IncreasingStartTime);
        }
    }
            
    public CepStream<StationData> Stations
    {
        get
        {
            return new CsvSequenceReaderTyped<StationData>
                (stationFile, true, StationData.GetFill())
                    .ToStream(Application, AdvanceTimeSettings.IncreasingStartTime)
                    .AlterEventLifetime(e => new DateTime(2000, 1, 1, 0, 0, 0, DateTimeKind.Utc),
                        e => TimeSpan.MaxValue);
        }
    }    
}

There are a few changes from the last section, to clean things up a bit in a project:

The assignment functions are returned by static functions on the data types themselves (as opposed to public properties – didn’t want to have a non-StreamInsight compatible data type in the event type definitions).
Instead of raising a generic wrapper class and then converting into PointEvent<T>, we instead raise TypedEvent<T> directly from the file (i.e. our enumerator is now IEnumerator<IntervalEvent<T>> GetEnumerator().
The file names are inferred to be in the same directory as the context assembly.
In the Stations stream, we perform an AlterEventLifetime to convert the list of stations into a reference stream by extending the time from January 1st 2000 (the timestamp of the first record in the data set), to max time.
- Ordinarily, this would be done with query logic in your application, but for ease of use in LINQPad (i.e. removing an extra step) I did it in the context.

Go ahead and compile the project, to obtain the WeatherData.LinqpadContext.dll. We’ll load this into LINQPad by

Add Connection, then select Microsoft StreamInsight from the list of drivers.
Select Custom Context from the Context Kind dropdown.
Click on the ellipsis for Assembly Path (...), then browse to the directory containing WeatherData.LinqpadContext.dll.
Click OK to finish selecting the new context.

Our weather context should now be available on the connections list. Before we can start querying these streams, we will need to take our data files (weather_data_out.csv.gz and station_data_out.csv.gz from our first step) and copy them to the assembly directory (in this case, the bin directory for our project).

If you’re not sure which directory you need to place the files in, simply run a query against the data context. The exception will show you the correct directory.

Now that we have our custom context ready to go, let’s run some queries.

Running some Queries

With the query context loaded up, let’s go ahead and run some queries against it.

From LINQPad, click File, New Query.
From the Language drop-down, select C# Statements.
From the Database drop-down, select StreamInsight: WeatherContext

Now we’re ready to run some queries. Let’s start with a couple of basic one to look at the SmallWeather and Stations streams.

Excellent! One last little query to close out our walkthrough of creating a data context – join the reference stream with the small data stream and filter for a specific airport.

Remember – you can execute just the selected text in LINQPad by highlighting it, then pressing F5 (or clicking Query –> Execute), which is why all of the screenshots show highlighted text.

Running ActiveMQ on Windows Azure: One Man’s Search for a Full Featured Messaging Broker Option in the Cloud

Christian.Martinez — Tue, 21 Dec 2010 20:24:00 GMT

I’ve always enjoyed building systems that utilized flexible messaging platforms. Before joining Microsoft, I was a member of a small ISV team that built a commercial .NET based ESB product (a rare creature in and of itself) that was primarily a WCF based messaging system. We had a number of capabilities that were useful and somewhat unique in their execution but one of my favorites was the ability to provide transactional messaging across multiple platforms by leveraging ActiveMQ in our adapter model.

I’ve always admired the Apache folks. More often than not they seem to just get it (unless you’re talking about docs … but hey their stuff is free and even us ginormous mega power companies screw the pooch on docs sometimes). And ActiveMQ is no exception. It has pretty much all of the capabilities you’d expect from a full featured broker including:

The list goes on. But you get it. It’s an actual message broker with tons of features, the most important of which is its .NET support via the .NET Messaging API.

There’s only one problem. ActiveMQ like pretty much every other messaging broker was developed pre-cloud. Granted, some of the AMQP brokers such as RabbitMQ have caught on with folks who make all of their money “off prem” but none of the brokers I’ve worked with are what I would call cloud friendly. Maybe there’s one out there and if there is please add a comment because I’d love to check it out.

So where does that leave us? Well if you’re just trying to get some things done in Windows Azure between roles and instances and you don’t need a full featured broker, stop right now and run to go see my colleague Valery Mizonov’s incredible blog about rolling your own. It’s the best thing I’ve seen on the subject and it was derived from real folks working on real challenges while deploying on Windows Azure. Valery is a genius though and he was built based on the design of the Cyberyne Systems Model 101 so it’s not a surprise that he was able to achieve such a masterpiece of engineering and bloggery.

If on the other hand you’re a simpler life form like me and feel a certain shall we say irritation that 58 billion in revenue and 14 billion in profit occurred without something more “out of the box” in the cloud for us messaging lovers then read on. But first, a warning:

WHAT YOU ARE ABOUT TO READ IS NOT IN ANY WAY A “BEST PRACTICE” OR EVEN SOMETHING NORMAL PEOPLE MIGHT TRY. IT IS AN ENVELOPE PUSHING EXERCISE THAT TURNED OUT PRETTY COOL AND LEFT ME WITH AN APPRECIATION OF THE POWER OF THE WINDOWS AZURE WORKER ROLE AND CONFIRMED MY APPRECIATION OF ACTIVEMQ.

Now, that the warning is out of the way let’s begin the journey…

Step 1: Figure Out How To Develop

I’d seen the posts on running Tomcat in Windows Azure and downloaded the Tomcat Accelerator. It gave me hope that what I wanted to do was possible. For what I wanted to do however, I wanted a more “integrated” experience than all of the cmd files in that project not to mention wanting to avoid having to build my own,.

So, I decided to use the technique I had used long ago as a Java Developer dealing with classpaths and a bazillion jar files: sheer brute force.

Yes, that is really is a JRE and ActiveMQ embedded with the files set to be copied into the project. It took a few monotonous minutes to set up but it was well worth it in the long run.

Step 2: Figure Out How Deal With Local Storage and Dynamic IP’s

Anyone who has spent time with Azure knows they need to deal with the rules of Local Storage and the way endpoints are assigned. When you’re dealing with something like starting a JVM and running something like ActiveMQ that has a lot of startup options then things get even more harrowing.

My first goal was then figuring out to get all the things I needed over to the Local Storage and to edit the config files so that when ActiveMQ started up the dynamically assigned ports were used.

This problem is ugly no matter how you slice it (the tomcat example I downloaded had robocopy in a .bat) but after I was done I didn’t feel too bad.

 private void ConfigureDirectories()
 {
            workingDir = Environment.GetEnvironmentVariable("RoleRoot") + @"\approot";
            javaHome = workingDir + @"\jre6";
            var activeMqFiles = workingDir + @"\activemq";
            runtimeDir = RoleEnvironment.GetLocalResource("Storage").RootPath + "activemq";
            Microsoft.VisualBasic.FileIO.FileSystem.CopyDirectory(activeMqFiles, runtimeDir, true);
  }

The awesomeness of the CLR shows up here as I leverage VB.NET for a little recursive copy assist.

Once everything was copied over then it was just a matter of setting up the environment and writing the IP’s to the respective config files

private void ConfigureEnvironment(Process proc)
{
            proc.StartInfo.EnvironmentVariables.Remove("JAVA_HOME");
            proc.StartInfo.EnvironmentVariables.Remove("ACTIVEMQ_HOME");
            proc.StartInfo.EnvironmentVariables.Add("JAVA_HOME", javaHome);
            proc.StartInfo.EnvironmentVariables.Add("ACTIVEMQ_HOME", runtimeDir);
            ManipulateIpAdresses();
 }

The rest of the set up is just Xpath boiler plate and Process set up so I won’t go into it here. One note, as of SDK 1.3 you have more options re: pinning the IP.

The important thing is after this part was done I had a broker firing up in dev fabric!

Step 3: Writing the Demo Code

This part was the easiest because the Apache guys have pretty much written everything hard which at the end of the day was the reason for trying to do this in the first place. I also wanted to try and use the durable and guaranteed features of ActiveMQ because there are already plenty of non durable, non guaranteed, multiple delivery options available without going to the pain of installing a 3rd part message broker.

My producer used the following simple driver logic and ran in one instance:

                    AttemptConnection();
                    using (var session = con.CreateSession(AcknowledgementMode.Transactional))
                    {
                        ITopic topic = session.GetTopic("AzureHostedTest");
                        var producer = session.CreateProducer(topic);
                        var start = DateTime.Now;
                        for (int i = 0; i < messageCount; i++)
                        {
                            ITextMessage msg = producer.CreateTextMessage("Hello world "+ msgNumber++);
                            producer.Send(msg);
                        }
                        session.Commit();
                        Thread.Sleep(sleepTime);
                    }

I just used the NMS libraries directly and didn’t use any of the Spring overlays.

My consumer logic was similarly simple

   var topic = new ActiveMQTopic("AzureHostedTest");
                        var mySession = session.CreateDurableConsumer(topic, con.ClientId, null, false);
                        mySession.Listener += (message) => {
                            ITextMessage msg = message as ITextMessage;
                            Trace.WriteLine(message.NMSMessageId + " " + msg.Text);
                            session.Commit();
                        };

Again, what I was aiming for was the durable options not perf. Like everything else with the product you have a ton of options in terms of perf tuning.

Much of the logic in the solution was concerned with attaining a connection. In a cloud world you need to be very concerned with things spinning up out of order or disappearing.

Ok. What are the odds of this running at this point. Pretty good actually, provided you wanted to use Local Storage as your durability option.

But we all know that’s just crazy. Nope you need something enterprise. Something serious. You need SQL Azure!

Step 4: Fighting Through the Topology Deployment Options

Once again ActiveMQ shines here in the features and flexibility department. However, this was also the part of the project where I knew the path that led me from being a pure open source guy to working for the evil empire was not out of naked greed or any malicious desire but was instead the need to get something @#$#$%#$$%^$ done! We have a lot of great products that let you actually do stuff. Sometimes the other nerd ecosystems out there seem to place productivity a distant second to coolness or “elegance” or their consultant friendly version of “enterprise” which results in a miserable experience for anyone outside of the tiny of circle of wonder nerds who grok/exploit their systems and style.

Oh, it looks like an innocuous enough exercise if you read this or this except for the pesky fact that if you use those pages as a guide with the latest ActiveMQ bits and the latest SQL Server JDBC drivers your chances of it working are nothing. Thus, I put myself into the mindset I use to live in 24/7 back in the old days and started to dig my way through bug lists and blogs and forums and experimentation until finally eureka! Or as I tend to think of it now for #$%#$%#$ sake!

Before sharing the tricks and tidbits however, let’s take a short break to describe the deployment I planned on. The topology I chose to deploy was what ActiveMQ calls JDBC Master Slave. This topology uses a sophisticated algorithm that basically says if I get the lock first then I’m the Master otherwise you are. I know I know. I should back off the rocket science a little bit here and make things a little easier to understand by means of a salient and powerful diagram!

My favorite part of this choice was this quote

“By default if you use the <jdbcPersistenceAdapter/> to avoid the high performance journal you will be using JDBC Master Slave by default. You just need to run more than one broker and point the client side URIs to them to get master/slave. This works because they both try an acquire an exclusive lock on a shared table in the database and only one will succeed”

Heck, it practically springs off the page as the right option. Now, before I go on I would be remiss if I didn’t mention the “high performance journal” referred to in the quote. In order to take advantage of them in Windows Azure however, you’re going to have to likely use a complex topology that works around the limitations of Windows Azure Drives and multiple instances. That’s the kind of thing done in our team’s Design Wins program though and is beyond the scope of this effort.

Besides, doing it this way you can use SQL Azure (of course in true open source fashion you could combine journals and JDBC..all you have to do is get through the docs etc ).

SQL Azure is an oasis of comfort and predictability in an otherwise turbulent world. It’s your solid and reliable friend in a cloud universe where we’re told that you might as well throw state out the window, abandon the thought of any guarantees and make everything idempotent because… well things are just going to happen in a way you don’t want them to ….so you need to be prepared. Granted, there’s some folks who have said people developing distributed systems should have been doing that all along even when they used on premises technologies, but just because they were right doesn’t mean we should have listened to them!

Now onto the secrets. After downloading the latest JDBC drivers I promptly waded through the xml docs and arrived at the following for my JDBC related settings:

 <persistenceAdapter>
      <jdbcPersistenceAdapter createTablesOnStartup="false"  dataDirectory="${activemq.base}/data" dataSource="#mssql-ds"/>
    </persistenceAdapter>

<bean id="mssql-ds" class="org.apache.commons.dbcp.BasicDataSource">
    <property name="driverClassName" value="com.microsoft.sqlserver.jdbc.SQLServerDriver"/>
    <property name="url" value="jdbc:sqlserver://[replace].database.windows.net;databaseName=ActiveMQ;"/>
    <property name="username" value="[replace]@[replace].database.windows.net"/>
    <property name="password" value="[replace]"/>
    <property name="maxActive" value="200"/>
    <property name="poolPreparedStatements" value="true"/>
  </bean>

I set createTablesOnStartup to true initially then false after they were created which seemed logical. Once you get past the unintuitive choice of class for datasource that’s not so bad. Unfortunately,when you start ActiveMQ that won’t work because you’ll run into bug similar to this one but with the added fun that the recommendations on the page don’t work for the 5.4.2 broker.

The answer to the issue is hinted at in this bug report. I’ll leave finding the hint to you gentle reader and skip right to its actual implementation because the hint will not help in isolation. You would need to carefully read the resulting stack trace after applying the hint and pair it with the knowledge gleaned to fully get the answer.

Here it is in a nutshell courtesy of 7-zip

See those files that are named slightly different than the ones that have the exact same content inside them as the highlighted ones. That’s the key. All you have to do is copy one of the files you need to base your new file on out of the jar, rename it using a slightly different convention, then stuff the renamed files back into the jar and away you go! What could be more natural?

Step 5: Deploy and Savor the Payoff

Once all the above was figured out all that was left was to grow a tree from an acorn while I waited for the project to upload to Windows Azure. Once it landed I monitored things with SSMS and a simple query

Then I got real brave and decided to cycle instance 0. Sure enough the pause in activity told me it was the master. When things picked up again after a few seconds I knew that the scruffy demo code had actually worked with failover. After that I forget because of the tears and the bourbon.

Here’s the code (be warned it’s fairly large because of the embedding).

Special thanks to Steve Marx, Paolo Salvatori and James Podgorski for their feedback and review.

Best Practices for Maximizing Scalability and Cost Effectiveness of Queue-Based Messaging Solutions on Windows Azure

Valery M — Mon, 20 Dec 2010 08:34:00 GMT

This post has been moved to a new location. Please follow the link below. Thank you.

http://windowsazurecat.com/2010/12/best-practices-for-maximizing-scalability-and-cost-effectiveness-of-queue-based-messaging-solutions-on-windows-azure/

How to use Duplex MEP to communicate with BizTalk from a .NET application or a WF workflow running inside AppFabric Part 2

Paolo Salvatori — Wed, 15 Dec 2010 05:00:00 GMT

Introduction

In the first article of the series we discussed how to exchange messages with an orchestration via a two-way WCF Receive Location using the Duplex Message Exchange Pattern. This form of bi-directional communication is characterized by the ability of both the service and the client to send messages to each other independently either using one-way or request/reply messaging. In a service-oriented architecture or a service bus composed of multiple, heterogeneous systems, interactions between autonomous client and service applications are asynchronous and loosely-coupled. All communications require published and discoverable service contracts, well-known data formats and a shared communication infrastructure. In this context, the use of asynchronous message patterns allows to increase the agility, scalability and flexibility of the overall architecture and helps decreasing the loose coupling of individual systems. In the second part of the article I’ll show you how to implement an asynchronous communication between a client application and a WCF Workflow Service running within IIS\AppFabric Hosting Services using the Durable Duplex Correlation provided by WF 4.0. Besides, I’ll demonstrate how to create a custom Activity for extending AppFabric Tracking with user-defined events and how to exploit the XML-based data transformation capabilities provided by the new BizTalk Server Mapper directly in a WF project thanks to the new Mapper Activity contained in the AppFabric Connect. The latter combines rich proven features of BizTalk Server 2010 with the flexible development experience of .NET to allow users to easily develop simple integration applications. Besides, AppFabric Connect allows you to extend the reach of your on-premise applications and services into Windows Azure AppFabric. In the future I’ll show you how to get advantage of the functionality offered by the AppFabric Connect to expose or move your BizTalk applications to the cloud using the Windows Azure AppFabric Service Bus. If you are interested in this subject, you can read the following articles:

“BizTalk AppFabric Connect: An Introduction" on the BizTalk Server Team Blog.
“Introducing BizTalk Server 2010 AppFabric Connect” article on TechNet.
“Exposing BizTalk Applications on the Cloud using AppFabric Connect for Services" article on TechNet.
"Exposing LOB Services on the Cloud Using AppFabric Connect for Services" article on TechNet

Before explaining the architecture of the demo, let me briefly introduce and discuss some of the techniques that I used to implement my solution.

Correlation in WF 4.0

Protocol-Based Correlation

For more information about protocol correlation, see the following topics on MSDN:

Content-Based Correlation

In my demo I used 2 different types of protocol-based correlation, respectively, the Request-Reply Correlation and the Durable Duplex Correlation. In the third part of this article I’ll show you how to use the Content-Based Correlation to implement an asynchronous communication between the WF workflow service and the underlying BizTalk orchestration.

AppFabric Monitoring and User-Defined Events

AppFabric provides new options and tools to monitor and troubleshoot the health of WCF and WF services running on IIS. The monitoring features support centralized event collection and analysis for WCF and WF services running on a single server. The monitoring features include the following:

A monitoring infrastructure that collects events from WCF and WF services and stores them in a Monitoring database.
A Monitoring database schema for instrumentation data. The Monitoring database stores tracked events from WCF and WF services in one unified data store.
A Windows Service called AppFabric Event Collection Service that, acting as an ETW consumer, collects and stores track events to the AppFabric Monitoring database.
An ApplicationServer module for Windows PowerShell that exposes monitoring cmdlets used to manage the Monitoring database and event collector sources.
An ApplicationServer module for Windows PowerShell that exposes tracing cmdlets, which you can use to configure tracing profiles, enable or disable tracing, and query trace logs.
A Monitoring Dashboard and other extensions to the IIS Manager console. You can use the Monitoring Dashboard to view selected metrics from the Monitoring database. You can use the IIS Manager extensions to manage monitoring databases, set the monitoring level, and query and analyze tracked events.

In a nutshell, here’s how AppFabric Monitoring works: event data is emitted from WCF and WF services and is sent to a high-performance Event Tracing for Windows (ETW) session. The data sent to an ETW session includes WCF analytic trace events and WF tracking record events emitted by using the ETW Tracking Participant. The AppFabric Event Collector Service harvests this event data from the above ETW session and stores this information in the Monitoring database. AppFabric monitoring tools can be used to analyze these events when they are persisted in the database. The AppFabric Monitoring features are fully documented on MSDN, so I will not cover this subject in detail. For more information on AppFabric Monitoring and ETW, see the following articles:

“Monitoring Architecture” topic on MSDN.
“Monitoring Applications” topic on MSDN.
“Windows Server AppFabric Dashboard Page” topic on MSDN.
“Workflow Tracking and Tracing” topic on MSDN.
“Improve Debugging And Performance Tuning With ETW” article on MSDN Magazine.

AppFabric Monitoring and Windows Workflow Tracking provide visibility into workflow execution. They provide the necessary infrastructure to track the execution of a workflow instance. The WF tracking infrastructure transparently instruments a workflow to emit records reflecting key events during the execution. In particular, AppFabric allows to configure, at a service level, a built-in or custom Tracking Profile to filter tracked data. Besides, WF provides the infrastructure and components to emit user-defined events to track custom application data. This brings us to the next topic.

Custom Activity for tracking User-Defined Events

While the base activity library includes a rich palette of activities for interacting with services, objects, and collections, it does not provide any activities for tracking user-defined events. Therefore, I decided to create a reusable WF custom activity to track user-defined events within any WF workflow service. This allows me to use the AppFabric Dashboard to analyze user events emitted at runtime by my WF services using my component. WF 4.0 provides a hierarchy of activity base classes from which you can choose from when building a custom activity. At a high level, the four base classes can be described as follows:

Activity – used to model activities by composing other activities, usually defined using XAML.
CodeActivity – a simplified base class when you need to write some code to get work done.
AsyncCodeActivity – used when your activity perform some work asynchronously.
NativeActivity – when your activity needs access to the runtime internals, for example to schedule other activities or create bookmarks

For more information on WF and custom activities, see the following article:

“A Developer's Introduction to Windows Workflow Foundation (WF) in .NET 4” by Matt Milner available.

The following table shows the code of my CustomTrackingActivity class.

#region Using Directives
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Diagnostics;
using System.Activities;
using System.Activities.Tracking;
using System.ComponentModel;
#endregion

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.WorkflowActivities
{
    /// <summary>
    /// This class can be used in any WF workflow to track
    /// user-defined events to any registered tracking providers
    /// </summary>
    [Designer(typeof(CustomTrackingActivityDesigner))]
    public sealed class CustomTrackingActivity : CodeActivity
    {
        #region Activity Arguments
        // Text Argument
        [DefaultValue(null)]
        public InArgument<string> Text { get; set; } 
        
        // TraceLevel Property
        public TraceLevel TraceLevel { get; set; }
        #endregion

        #region Protected Methods
        /// <summary>
        /// Tracks the text message contained in the Text argument.
        /// </summary>
        /// <param name="context">The execution context under which the activity executes.</param>
        protected override void Execute(CodeActivityContext context)
        {
            // Obtain the runtime value of the Text and TraceLevel input arguments
            string text = context.GetValue(this.Text);

            // Create and initialize a custom tracking record 
            CustomTrackingRecord record = new CustomTrackingRecord(text, this.TraceLevel);

            // Sends the specified custom tracking record to any registered tracking providers
            context.Track(record);
        } 
        #endregion
    }
}

The CustomTrackingActivity class is derived from the CodeActivity base class and overrides the Execute. The latter has a single parameter of type CodeActivityContext which represents the execution context under which the activity executes. In particular, the context object exposes a method called Track that can be used to send a custom tracking record to any registered tracking providers. The tracking provider used by AppFabric is the EtwTrackingParticipant. For more information, see the following article:

“Tracking Events into Event Tracing in Windows” on MSDN.

The CustomTrackingActivity class exposes 2 properties:

Text: contains the message to track at runtime. This latter can be declaratively set using Workflow Designer.
TraceLevel: specifies the intended trace level for the message.

The code of the custom activity is very straightforward and self-explaining. First, the method invokes the GetValue method exposed by the context to retrieve the value of the Text property, next it creates a new instance of the CustomTrackingRecord class and finally it calls the Track method on the context object to track the user-defined event. This activity can surely be extended to extract and track business relevant data associated with the workflow variables.

To control the look and feel of the custom Activity, I added an Activity Designer item to my project. In particular, this allowed me to perform the following customizations:

Specify a custom icon in the top-left corner of the Activity
Create an ExpressionTextBox control and bind it to the Text property of the custom activity.
Create a ComboBox control and bind it to the TraceLevel property of the custom activity.

Note
To associate the Activity Designer with the custom activity I decorated this with a DesignerAttribute and I specified the type of the Designer class as argument.

The following table contains the XAML code for the designer. I’m certainly not a WPF expert, so even if there’s probably a better way to achieve the same result, the code below perfectly fits my needs.

<sap:ActivityDesigner x:Class="Microsoft.AppFabric.CAT.Samples.DuplexMEP.WorkflowActivities.CustomTrackingActivityDesigner"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:sys="clr-namespace:System;assembly=mscorlib"
    xmlns:diag="clr-namespace:System.Diagnostics;assembly=system"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    xmlns:sap="clr-namespace:System.Activities.Presentation;assembly=System.Activities.Presentation"
    xmlns:sapv="clr-namespace:System.Activities.Presentation.View;assembly=System.Activities.Presentation">
    <sap:ActivityDesigner.Resources>
        <ObjectDataProvider MethodName="GetValues"
                        ObjectType="{x:Type sys:Enum}"
                        x:Key="TraceLevelValues">
            <ObjectDataProvider.MethodParameters>
                <x:Type TypeName="diag:TraceLevel" />
            </ObjectDataProvider.MethodParameters>
        </ObjectDataProvider>
    </sap:ActivityDesigner.Resources>
    <sap:ActivityDesigner.Icon>
        <DrawingBrush>
            <DrawingBrush.Drawing>
                <ImageDrawing>
                    <ImageDrawing.Rect>
                        <Rect Location="0,0" Size="25,25" ></Rect>
                    </ImageDrawing.Rect>
                    <ImageDrawing.ImageSource>
                        <BitmapImage UriSource="Resources/ActivityIcon.gif"/>
                    </ImageDrawing.ImageSource>
                </ImageDrawing>
            </DrawingBrush.Drawing>
        </DrawingBrush>
    </sap:ActivityDesigner.Icon>
    <Grid Margin="10">
        <Grid.RowDefinitions>
            <RowDefinition Height="30"/>
            <RowDefinition Height="30"/>
        </Grid.RowDefinitions>
        <Grid.ColumnDefinitions>
            <ColumnDefinition/>
            <ColumnDefinition/>
        </Grid.ColumnDefinitions>
        <Label Grid.Row="0" 
               Grid.Column="0"
               VerticalAlignment="Center">Text:</Label>
        <sapv:ExpressionTextBox Grid.Row="0" 
                                Grid.Column="1"
                                x:Name="expText"
                                OwnerActivity="{Binding Path=ModelItem, Mode=TwoWay}"
                                Expression="{Binding Path=ModelItem.Text.Expression, Mode=TwoWay}"
                                ExpressionType="{x:Type TypeName=sys:String}"
                                HintText="Message Text"
                                VerticalAlignment="Center"
                              />
        <Label Grid.Row="1" 
               Grid.Column="0"
               VerticalAlignment="Center">TraceLevel:</Label>
        <ComboBox Grid.Row="1" 
                  Grid.Column="1"
                  VerticalAlignment="Center"
                  Name="myComboBox" 
                  ItemsSource="{Binding Source={StaticResource TraceLevelValues}}"
                  SelectedValue="{Binding Path=ModelItem.TraceLevel, Mode=TwoWay}"/>
    </Grid>
</sap:ActivityDesigner>

To develop my custom activity, I used the following resources:

“Designing and Implementing Custom Activities” topic on MSDN.
“Using Custom Activity Designers and Templates” topic on MSDN.
“A Developer's Introduction to Windows Workflow Foundation (WF) in .NET 4” by Matt Milner available.
“From Code Activity to Custom Activity” video on Channel9.
“WF 4.0 Custom Activities Part 3 Activity Designers” video in CloudCasts site.
“WF 4.0 Custom Activity Designer” post on Sreeni.NET Blog.
“The Workflow 4.0 Custom Activity Icon (and Toolbox Icon) FAQ“ post on the Activity Designer blog.

If you are interested in how to track user-defined events in a WCF service running within AppFabric, you can review the following article:

“Getting the most out of user events and Windows Server AppFabric Monitoring” by Emil Velinov on the AppFabric CAT blog.

Transforming Messages within a WF Workflow using the Mapper Activity

As you will see in the next section, my demo is composed of 3 major tiers:

A Windows Forms client application.
A WCF Workflow Service running in Windows Server AppFabric.
A BizTalk Application composed of an orchestration and a request-response WCF Receive Location.

In a few words, the WCF Workflow Service receives a request from the client application and invokes the downstream orchestration via a WCF Receive Location. The format of the request and response messages exchanged by the WCF workflow service, respectively, with the client and BizTalk applications is obviously different to reflect real-world EAI scenarios where heterogeneous systems use different schemas to represent the same entities. Now, in this case, message transformation could be configured in a declarative way to run on the BizTalk receive location that receives requests and returns related responses. Nevertheless, since one of the objectives of this article was introducing the new Mapper activity available in AppFabric Connect, I decided to implement message transformation within my WCF workflow service. AppFabric Connect provides WF developers access to both the BizTalk Mapper and the BizTalk Adapter Pack 2010. Utilizing the Mapper activity, developers can exploit the features supplied by the BizTalk Server 2010 Mapper to design and use transformation maps in a WF workflow service hosted in IIS\AppFabric.

Using the Mapper activity in a WF workflow is quite straightforward: first of all, you have to define the WCF data contract classes that model the source and destination messages. Thus, in my solution I created 5 data contract classes to model the messages exchanged by WCF workflow service, respectively, with the client and BizTalk applications:

WFRequest: defines the request message sent by the client application to the WCF workflow service.
WFAck: specifies the structure of the acknowledgement message sent by the WCF workflow service to the client application.
WFResponse: represents the the response message returned by the WCF workflow service to the client application.
BizTalkRequest: defines the request message sent by the WCF workflow service to the BizTalk application.
BizTalkResponse: represents the the response message returned by the BizTalk application to the WCF workflow service

These 5 data classes belong to the same Class Library project called DataContracts. For your convenience, below I included the code of these components.

WFRequest Class

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.DataContracts
{
    [DataContract(Name="Request", Namespace="http://microsoft.appfabric.cat/10/samples/duplexmep/wf")]
    public class WFRequest : IExtensibleDataObject
    {
        #region Private Fields
        private ExtensionDataObject extensionData;
        private string id;
        private string question;
        private int delay;
        #endregion

        #region Public Constructors
        public WFRequest()
        {
            this.id = null;
            this.question = null;
            this.delay = 0;
        }

        public WFRequest(string question, int delay)
        {
            this.id = Guid.NewGuid().ToString();
            this.question = question;
            this.delay = delay;
        }
        #endregion

        #region Public Properties
        public ExtensionDataObject ExtensionData
        {
            get
            {
                return this.extensionData;
            }
            set
            {
                this.extensionData = value;
            }
        }

        [DataMember(Name = "Id", IsRequired = true, Order = 1, EmitDefaultValue = true)]
        public string Id
        {
            get
            {
                return this.id;
            }
            set
            {
                this.id = value;
            }
        }

        [DataMember(Name = "Question", IsRequired = true, Order = 2, EmitDefaultValue = true)]
        public string Question
        {
            get
            {
                return this.question;
            }
            set
            {
                this.question = value;
            }
        }

        [DataMember(Name = "Delay", IsRequired = true, Order = 3, EmitDefaultValue = true)]
        public int Delay
        {
            get
            {
                return this.delay;
            }
            set
            {
                this.delay = value;
            }
        }
        #endregion
    }
}

WFAck

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.DataContracts
{
    [DataContract(Name = "Ack", Namespace = "http://microsoft.appfabric.cat/10/samples/duplexmep/wf")]
    public class WFAck : IExtensibleDataObject
    {
        #region Private Fields
        private ExtensionDataObject extensionData;
        private string id;
        private string ack;
        #endregion

        #region Public Constructors
        public WFAck()
        {
            this.id = null;
            this.ack = null;
        }

        public WFAck(string id, string question)
        {
            this.id = id;
            this.ack = question;
        }
        #endregion

        #region Public Properties
        public ExtensionDataObject ExtensionData
        {
            get
            {
                return this.extensionData;
            }
            set
            {
                this.extensionData = value;
            }
        }

        [DataMember(Name = "Id", IsRequired = true, Order = 1, EmitDefaultValue = true)]
        public string Id
        {
            get
            {
                return this.id;
            }
            set
            {
                this.id = value;
            }
        }

        [DataMember(Name = "Ack", IsRequired = true, Order = 2, EmitDefaultValue = true)]
        public string Ack
        {
            get
            {
                return this.ack;
            }
            set
            {
                this.ack = value;
            }
        }
        #endregion
    }
}

WFResponse

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.DataContracts
{
    [DataContract(Name = "Response", Namespace = "http://microsoft.appfabric.cat/10/samples/duplexmep/wf")]
    public class WFResponse : IExtensibleDataObject
    {
        #region Private Fields
        private ExtensionDataObject extensionData;
        private string id;
        private string answer;
        #endregion

        #region Public Constructors
        public WFResponse()
        {
            this.id = null;
            this.answer = null;
        }

        public WFResponse(string id, string question)
        {
            this.id = id;
            this.answer = question;
        }
        #endregion

        #region Public Properties
        public ExtensionDataObject ExtensionData
        {
            get
            {
                return this.extensionData;
            }
            set
            {
                this.extensionData = value;
            }
        }

        [DataMember(Name = "Id", IsRequired = true, Order = 1, EmitDefaultValue = true)]
        public string Id
        {
            get
            {
                return this.id;
            }
            set
            {
                this.id = value;
            }
        }

        [DataMember(Name = "Answer", IsRequired = true, Order = 2, EmitDefaultValue = true)]
        public string Answer
        {
            get
            {
                return this.answer;
            }
            set
            {
                this.answer = value;
            }
        }
        #endregion
    }
}

BizTalkRequest

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.DataContracts
{
    [DataContract(Name="Request", Namespace="http://microsoft.appfabric.cat/10/samples/duplexmep")]
    public partial class BizTalkRequest : IExtensibleDataObject
    {
        #region Private Fields
        private ExtensionDataObject extensionData;
        private string id;
        private string question;
        private int delay;
        #endregion

        #region Public Constructors
        public BizTalkRequest()
        {
            this.id = Guid.NewGuid().ToString();
            this.question = null;
            this.delay = 0;
        }

        public BizTalkRequest(string question, int delay)
        {
            this.id = Guid.NewGuid().ToString();
            this.question = question;
            this.delay = delay;
        }
        #endregion

        #region Public Properties
        public ExtensionDataObject ExtensionData
        {
            get
            {
                return this.extensionData;
            }
            set
            {
                this.extensionData = value;
            }
        }

        [DataMemberAttribute(IsRequired = true, Order = 1)]
        public string Id
        {
            get
            {
                return this.id;
            }
            set
            {
                this.id = value;
            }
        }

        [DataMemberAttribute(IsRequired = true, Order = 2)]
        public string Question
        {
            get
            {
                return this.question;
            }
            set
            {
                this.question = value;
            }
        }

        [DataMemberAttribute(IsRequired = true, Order = 3)]
        public int Delay
        {
            get
            {
                return this.delay;
            }
            set
            {
                this.delay = value;
            }
        } 
        #endregion
    }
}

BizTalkResponse

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.DataContracts
{
    [DataContract(Name = "Response", Namespace = "http://microsoft.appfabric.cat/10/samples/duplexmep")]
    public partial class BizTalkResponse : IExtensibleDataObject
    {
        #region Private Fields
        private ExtensionDataObject extensionData;
        private string id;
        private string answer;
        #endregion

        #region Public Properties
        public ExtensionDataObject ExtensionData
        {
            get
            {
                return this.extensionData;
            }
            set
            {
                this.extensionData = value;
            }
        }

        [DataMemberAttribute(IsRequired = true, Order = 1)]
        public string Id
        {
            get
            {
                return this.id;
            }
            set
            {
                this.id = value;
            }
        }

        [DataMemberAttribute(IsRequired = true, Order = 2)]
        public string Answer
        {
            get
            {
                return this.answer;
            }
            set
            {
                this.answer = value;
            }
        }
        #endregion
    }
}

The next step was to define a variable for each of the messages exchanged by the WCF workflow service with the client and the BizTalk application. The picture below shows the 5 data contract variables that I created within the outermost Sequential activity within my WCF workflow service. This activity contains also the correlation handle used by the workflow to correlate the request message sent by the client application with the corresponding response message. We will expand on this point later.

Then I created a map to transform a WFRequest object into a BizTalkRequest object and another map to transform a BizTalkResponse object into an instance of the WFResponse class. In a nutshell, these are the steps I followed to create the first of the 2 transformation maps. After installing the BizTalk Server 2010 developer tools and the WCF LOB Adapter SDK, you can see the Mapper activity on the Windows Workflow Activity Palette under a tab called BizTalk, shown in the picture below.

When you drag the Mapper activity onto a workflow, it prompts you for the data types of the source and destination message. The dialog allows you to choose primitive types, or custom types. To create the first map, I chose the WFRequest type as InputDataContractType and the BizTalkRequest as OutputDataContractType as shown in the picture below.

When I clicked the OK button, an un-configured Mapper activity appeared in my workflow. After setting the explicit names of my source (WFRequest) and destination variables (BizTalkRequest) in the Mapper activity’s Property window, I clicked the Edit button and chose to create a new map.

The Select a map dialog allows to create a new map or select an existing one based on data contract types chosen at the previous step. If you are creating a new map, the activity will generate the XML schemas for the selected input and output data contract types and a new BizTalk map (.btm) file. Differently than in BizTalk Server, XML schemas and map files don’t need to published to a centralized database, but they become an integral part of the project. You can eventually define your maps in a separate project from your workflows to increase the reusability and maintenance level of these artifacts.

Upon clicking the OK button, the BizTalk Mapper Designer appeared and I could create my transformation map as shown in the picture below. When using the BizTalk Mapper Designer within a WF workflow application, you have full access to all the features and functoids that BizTalk developers normally use in their solution.

At runtime, the input data is first serialized into XML and then transformed using the XSLT generated from the map file. The message resulting from the transformation is finally de-serialized back into an object of the output type. At this regard, the Mapper activity decides to use the XmlSerializer only if the type is annotated with the XmlTypeAttribute or XmlRootAttribute. If the activity is an array type, the attribute check will be performed on the array element type as well. In all other cases, DataContractSerializer is used.

Note
When creating a map it is possible to use Advanced Options to change the Serializer to use for Input and the Serializer to use for Result from Auto to DataContractSerializer or XMLSerializer. Manually setting the serializer is not recommended however and extreme caution should be exercised when doing this because specifying the wrong serializer will cause serialization to fail at runtime.

For more information on the Mapper Activity, see the following articles by Trace Young:

”BizTalk Server 2010 AppFabric Connect feature Frequently Asked Questions (FAQ)” article on TechNet.
”Installing the BizTalk Server 2010 AppFabric Connect feature” article on TechNet.
“Using the New BizTalk Mapper Shape in a Windows Workflow Service” article on Richard Seroter’s blog.

After stacking the necessary (LEGO) bricks, we are now ready to build our solution, so let’s get started.

Using a Durable Duplex Correlation to communicate with a WF Workflow

The following picture depicts the architecture of the use case. The idea behind the application is quite straightforward: a Windows Forms application submits a question to a WCF workflow service hosted in IIS\AppFabric and asynchronously waits for the related answer. The WCF workflow service uses the Mapper activity to transform the incoming request in a format suitable to be consumed by the underlying BizTalk application and synchronously invokes the SyncMagic8Ball orchestration via a WCF-NetTcp Receive Location. The orchestration is a BizTalk version of the notorious Magic 8 Ball toy and it randomly returns one of 20 standardized answers. Upon receiving the response message from BizTalk, the WCF workflow service applies another map using the Mapper activity and returns the resulting message to the client application. In this version, the client application communicates with the WCF workflow service using a pattern called Durable Duplex Correlation, whereas the WCF workflow service communicates with the BizTalk application using a synchronous message exchange pattern. In the next and final article of the series, I’ll show you how to use the Content-Based Correlation to implement an asynchronous message exchange between the WCF workflow service and the downstream BizTalk orchestration.

Message Flow

The Windows Forms Client Application enables a user to specify a question and a delay in seconds. When the user presses the Ask button, a new request message containing the question and the delay is created and sent to a the WCF workflow service. Before sending the first message, the client application creates and opens a service host to expose a callback endpoint that the workflow can invoke to return the response message. In particular, the binding used to expose this callback contract is the NetTcpBinding, whereas the binding used to send the request message to the WCF workflow service is the NetTcpContextBinding. We will expand on this point in the next sections when we’ll analyze the client-side code.
The WCF workflow service receives the request message of type WFRequest using the Receive activity of a ReceiveAndSendReply composed activity. Then it uses this activity to initialize the callback correlation handle.
The WCF workflow service uses the CustomTrackingActivity to keep track of individual processing steps and uses an instance of the Mapper activity to transform the WFRequest object into an instance of the BizTalkRequest class.
The WCF workflow service uses a WCF proxy activity to send the BizTalkRequest message to the WCF-NetTcp receive location exposed by the BizTalk application.
The WCF receive location receives the request message and the XmlReceive pipeline promotes the MessageType context property.
The Message Agent submits the request message to the MessageBox (BizTalkMsgBoxDb).
A new instance of the SyncMagic8Ball orchestration receives the request message via a two-way logical port and uses a custom helper component called XPathHelper to read the value of the Question and Delay elements from the inbound message.
The SyncMagic8Ball orchestration invokes the SetResponse static method exposed by the ResponseHelper class to build the response message containing the answer to this question contained in the request message. The response message is then published to the MessageBox (BizTalkMsgBoxDb) by the Message Agent.
The response message is retrieved by the WCF-NetTcp Receive Location.
The PassThruTransmit send pipeline is executed by the WCF-NetTcp Receive Location.
The response message is returned to the WCF workflow service.
The WCF workflow service uses a Mapper activity to transform the BizTalkResponse object into an instance of the WFRequest class.
The WCF workflow service uses a Send activity to send back the response message to the client application. The Send activity is configured to use the callback correlation that contains the URI of the callback endpoint exposed by the client application.

Durable Duplex

One of the objective of this post is demonstrating how to use a duplex communication pattern to exchange messages between a client application and a WCF workflow service running within IIS\AppFabric. However, WF 4.0 doesn't directly support WCF Duplex communication, but it supports a different pattern called Durable Duplex. This pattern requires the client and server applications to use a separate WCF channel to exchange, respectively, the request and response message and this enables them to use a different binding on the callback channel than the one used to send the original request. Since the channels used to exchange the request and response message are independent, the callback can happen at any time in the future. The only requirement for the caller is to have an active endpoint listening for the callback message. The Durable Duplex pattern allows a client application to communicate with a WCF workflow service in a long-running conversation.
To use durable duplex correlation, the client application and the WCF workflow service must use a context-enabled binding that supports two-way operations, such as NetTcpContextBinding or WSHttpContextBinding. This requirement applies only to the WCF channel used to exchange the initial request, whereas any binding can be used by the WCF channel used by the callback. Before sending a request message, the client application registers a ClientCallbackAddress with the URI of the callback Endpoint. The WCF workflow service receives this data with a Receive activity and then uses it on its own Endpoint in the Send activity to send a response or a notification message back to the caller. For more information on the Durable Duplex Correlation, see the following topic on MSDN:

“Durable Duplex Correlation” topic in the WF online documentation.

Client Code

The following table contains code used by the client application to invoke the WCF workflow service using the Durable Duplex communication pattern.

private void btnAsk_Click(object sender, EventArgs e)
{
    try
    {
        int delay = 0;
        //  Question Validation
        if (string.IsNullOrEmpty(txtQuestion.Text))
        {
            WriteToLog(QuestionCannotBeNull);
            txtQuestion.Focus();
            return;
        }
        //  Delay Validation
        if (string.IsNullOrEmpty(txtDelay.Text) ||
            !int.TryParse(txtDelay.Text, out delay))
        {
            WriteToLog(DelayMustBeANumber);
            txtDelay.Focus();
            return;
        }
        // Endpoint Validation
        if (string.IsNullOrEmpty(cboEndpoint.Text))
        {
            WriteToLog(NoEndpointsFound);
        }

        if (serviceHost == null)
        {
            try
            {
                // Find a free TCP port
                int port = FreeTcpPort();
                // Set the value of the static MainForm property
                Magic8BallWFCallback.MainForm = this;
                // Create the service host that will be used to 
                // receive the response from the WCF Workflow Service
                serviceHost = new ServiceHost(typeof(Magic8BallWFCallback));
                if (serviceHost.Description.Endpoints.Count > 0)
                {
                    // Read the URI from the configuration file and
                    // change it to use the TCP port found at the previous step
                    Uri oldUri = serviceHost.Description.Endpoints[0].Address.Uri;
                    listenUri = new Uri(string.Format(URIFormat, oldUri.Scheme, 
                                                        oldUri.Host, port, 
                                                        oldUri.AbsolutePath));
                    serviceHost.Description.Endpoints[0].Address = new EndpointAddress(listenUri);
                    serviceHost.Open();
                    // Log the URI
                    WriteToLog(string.Format(Magic8BallWFCallbackServiceOpened, listenUri.AbsoluteUri));
                }
                else
                {
                    // Log error message
                    WriteToLog(NoValidEndpointsForMagic8BallWFCallbackServiceOpened);
                }
            }
            catch (Exception ex)
            {
                // Log Exception and InnerException
                WriteToLog(ex.Message);
                WriteToLog(ex.InnerException.Message);
            }
        }
        Magic8BallWFClient proxy = null;
        try
        {
            if (serviceHost != null && serviceHost.State == CommunicationState.Opened)
            {
                // Create the client proxy to send the question to the WCF Workflow Service
                proxy = new Magic8BallWFClient(cboEndpoint.Text);
                // Create a new request message
                WFRequest request = new WFRequest();
                request.Id = Guid.NewGuid().ToString();
                request.Question = txtQuestion.Text;
                request.Delay = delay;
                WriteToLog(string.Format(CultureInfo.CurrentCulture, 
                                            RequestFormat, 
                                            cboEndpoint.Text, 
                                            request.Id, 
                                            request.Question));
                using (new OperationContextScope((IContextChannel)proxy.InnerChannel))
                {
                    // You can use the context to send pairs of keys and values, 
                    // stored implicitly in the message headers,
                    IDictionary<string, string> dictionary = new Dictionary<string, string>();
                    dictionary["MachineName"] = Environment.MachineName;
                    
                    // Add the URI of the callback endpoint to the callback context
                    // This information is used by the WCF workflow service to initialize
                    // the callback correlation handle used to implement the Durable Duplex pattern.
                    var context = new CallbackContextMessageProperty(listenUri, dictionary);
                    OperationContext.Current.OutgoingMessageProperties.Add(CallbackContextMessageProperty.Name,
                                                                           context);
                    // Invoke the WF Workflow Service
                    WFAck ack = proxy.AskQuestion(request);
                    if (ack != null &&
                        !string.IsNullOrEmpty(ack.Ack))
                    {
                        WriteToLog(string.Format(AckFormat, ack.Id, ack.Ack));
                    }
                }
            }
        }
        catch (FaultException ex)
        {
            WriteToLog(ex.Message);
            if (proxy != null)
            {
                proxy.Abort();
            }
        }
        catch (CommunicationException ex)
        {
            WriteToLog(ex.Message);
            if (proxy != null)
            {
                proxy.Abort();
            }
        }
        catch (TimeoutException ex)
        {
            WriteToLog(ex.Message);
            if (proxy != null)
            {
                proxy.Abort();
            }
        }
        catch (Exception ex)
        {
            WriteToLog(ex.Message);
            if (proxy != null)
            {
                proxy.Abort();
            }
        }
    }
    catch (Exception ex)
    {
        WriteToLog(ex.Message);
    }
}

In detail, the code performs the following steps:

Upon the dispatch of the first message, the client application creates the service host that will be used to asynchronously receive the response message from the WCF workflow service. In particular, it individuates a free TCP port, then reads the configuration of the callback service endpoint from the configuration file and finally modifies the URI of the endpoint to use the TCP port found. The WCF callback service exposed by the client application is defined by the Magic8BallWFCallback class that implements the IMagic8BallWFCallback service interface.
Creates a new instance of the Magic8BallWFClient proxy class.
Creates and initializes a new WFRequest object.
Uses the OperationContextScope to access the OperationContext within a scope.
Uses the current OperationContext and an object of type CallbackContextMessageProperty to initialize the wsc:CallbackContext message header with the URI of the callback service endpoint.
Invokes the WCF workflow service.

The following table reports the code of the Magic8BallWFCallback service class contained in the Client project and the code of the IMagic8BallWFCallback service interface contained in the ServiceContracts project.

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.ServiceContracts
{
    [ServiceContract(Namespace = http://microsoft.appfabric.cat/10/samples/duplexmep/wf, 
                     ConfigurationName = "IMagic8BallWFCallback")]
    public interface IMagic8BallWFCallback
    {
        [OperationContract(Action = "AskQuestionResponse", IsOneWay = true)]
        void AskQuestionResponse(WFResponseMessage responseMessage);
    }
}

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.Client
{
    [ServiceBehavior]
    public class Magic8BallWFCallback : IMagic8BallWFCallback
    {
        #region Private Constants
        private const string ResponseFormat = "Response:\n\tId: {0}\n\tAnswer: {1}";
        #endregion

        #region Private Static Fields
        private static MainForm form = null;
        #endregion

        #region Private Static Fields
        public static MainForm MainForm
        {
            get
            {
                return form;
            }
            set
            {
                form = value;
            }
        }
        #endregion

        #region IMagic8BallCallback Members
        [OperationBehavior]
        public void AskQuestionResponse(WFResponseMessage responseMessage)
        {
            if (responseMessage != null &&
                responseMessage.Response != null &&
                !string.IsNullOrEmpty(responseMessage.Response.Id) &&
                !string.IsNullOrEmpty(responseMessage.Response.Answer))
            {
                form.WriteToLog(string.Format(CultureInfo.CurrentCulture, 
                                              ResponseFormat, 
                                              responseMessage.Response.Id, 
                                              responseMessage.Response.Answer));
            }
        }
        #endregion
    }
}

namespace Microsoft.AppFabric.CAT.Samples.DuplexMEP.Client
{
    public partial class MainForm : Form
    {
        ...
        public void WriteToLog(string message)
        {
            if (InvokeRequired)
            {
                Invoke(new Action<string>(InternalWriteToLog), new object[] { message });
            }
            else
            {
                InternalWriteToLog(message);
            }
        }

        private void InternalWriteToLog(string message)
        {
            if (message != null &&
                message != string.Empty)
            {
                string[] lines = message.Split('\n');
                DateTime objNow = DateTime.Now;
                string space = new string(' ', 19);
                string line;

                for (int i = 0; i < lines.Length; i++)
                {
                    if (i == 0)
                    {
                        line = string.Format(DateFormat,
                                             objNow.Hour,
                                             objNow.Minute,
                                             objNow.Second,
                                             lines[i]);
                        lstLog.Items.Add(line);
                    }
                    else
                    {
                        lstLog.Items.Add(space + lines[i]);
                    }
                }
                lstLog.SelectedIndex = lstLog.Items.Count - 1;
            }
        }
        ...
    }
}

The AskQuestionResponse operation exposed by the callback contract validates and logs the content of the response message returned by the WCF workflow service. The following table shows the content of the configuration file of the client application.

<?xml version="1.0"?>
<configuration>
  <system.serviceModel>
    <bindings>
      <netTcpBinding>
        <binding name="netTcpBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpBinding>
      <netTcpContextBinding>
        <binding name="netTcpContextBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpContextBinding>
    </bindings>
    <services>
      <service name="Microsoft.AppFabric.CAT.Samples.DuplexMEP.Client.Magic8BallWFCallback">
        <endpoint address="net.tcp://localhost:15001/magic8ballwfcallback"
                  binding="netTcpBinding"
                  bindingConfiguration="netTcpBinding"
                  contract="IMagic8BallWFCallback"/>
      </service>
    </services>
    <client>
      <clear />
      <endpoint address="net.tcp://localhost/magic8ballwf/syncmagic8ball.xamlx"
                binding="netTcpContextBinding" 
                bindingConfiguration="netTcpContextBinding"
                contract="IMagic8BallWF"
                name="NetTcpEndpointSyncWF" />
    </client>
  </system.serviceModel>
  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.0"/>
  </startup>
</configuration>

As you can easily notice, the client endpoint uses the NetTcpContextBinding, whereas the callback service endpoint uses the NetTcpBinding. The context bindings were introduced in the .NET Framework 3.5 and are typically used the same way as their base bindings. However, they add support for a dedicated context management protocol. These bindings can be used with or without a context. The context protocol lets you pass as a custom context a collection of strings in the form of key-value pairs, stored implicitly in the message headers. In our context, the use of a context binding is mandatory to initialize the Durable Duplex Correlation. This brings us to the next topic.

WCF Workflow Service

WCF workflow services provide a productive environment for authoring long-running, durable operations or services. Workflow services are implemented using WF activities that can make use of WCF for sending and receiving data. Explaining in detail how to build a WCF workflow service is out of the scope of the present article. For more information on WCF workflow services, see the following articles:

“How to: Create a Workflow Service with Messaging Activities” topic on MSDN.
“Visual Design of Workflows with WCF and WF 4” article on MSDN Magazine.
“Building WCF Workflow Services and Hosting in AppFabric” post on Richard Seroter’s blog.

In this section I will focus my attention on how the WCF workflow service implements communications with both the client application and the BizTalk orchestration. When I created the WCF Workflow Service, the initial workflow just contained a Sequence activity with a Receive activity followed by a SendReply activity as shown in the following illustration.I selected the Sequential activity and I clicked the Variables button to display the corresponding editor. I created a variable for each message to exchange with the client and BizTalk application and then I created a CorrelationHandle variable to hold the callback correlation.

In order to expose a NetTcpContextBinding endpoint I configured the Receive activity as shown in the following picture:

On the left panel of the dialog, I selected the correlation handle variable previously created, and I then chose Callback correlation initializer in the combobox containing the available correlation type initializers. Before invoking the downstream BizTalk application, the WCF workflow service immediately returns an ACK message to the caller. Therefore, I configured the SendReply activity, bound to the initial Receive activity, to return a WFAck message, as shown in the picture below.

At this point, I had two option to invoke the WCF-NetTcp receive location exposed by the BizTalk application : the first choice was generating a custom WCF proxy activity, whereas the second alternative was using the messaging activities provided out-of-the-box by WF. In this case I opted for the first option, but in the next article I’ll show you how using the messaging activities and the Content-Based Correlation to implement an asynchronous communication between the WCF workflow service and the underlying orchestration. To create the WCF proxy activity I performed the following steps:

I right clicked the WorkflowServices project and I selected the Add Service Reference from the context menu.
As shown in the following picture, in the Add Service Reference dialog I specified the URL of the MetadataExchange endpoint exposed by the DuplexMEP.Sync.WCF-NetTcp.ReceiveLocation. Previously, I generated the service metadata endpoint using the BizTalk WCF Publishing Wizard. This has been recently extended in AppFabric Connect to provide the possibility to publish on-premise WCF receive locations to the cloud using the Windows Azure Service Bus. For more information on this topic, see “Exposing BizTalk Applications on the Cloud using AppFabric Connect for Services" on TechNet.
Then I specified a namespace for the proxy activity and I clicked the OK button to confirm.

In general, this operation generates a custom activity for each operation exposed by the referenced service. As shown in the picture below, in my case this action created a single activity named AskQuestion to invoke the request-response WCF receive location exposed by the BizTalk application.

The following picture depicts the central part of the SyncMagic8Ball WCF workflow service.

This section of the workflow executes the following actions:

Tracks a user-defined event using the CustomTrackingActivity.
Uses the Mapper activity to transform the WFRequest message into a BizTalkRequest message.
Tracks a user-defined event using the CustomTrackingActivity.
Invokes the WCF receive location exposed by the BizTalk application.
Tracks a user-defined event using the CustomTrackingActivity.
Uses the Mapper activity to transform the BizTalkResponse message into a WFResponse message.
Tracks a user-defined event using the CustomTrackingActivity.

This portion of the workflow performs just 2 steps:

Uses a Send activity to send the response message to the caller. This activity is configured to use the callback handle correlation.
Tracks a user-defined event using the CustomTrackingActivity.

The following figure shows how I configured the Send activity used to transmit the response message back to the caller.

As highlighted above, I assigned to the CorrelatesWith property the callback correlation handle that I previously initialized on the Receive activity. Then I properly set the other properties like OperationName, Action, and ServiceContractName to match the characteristics of the callback service endpoint exposed by the client application.

BizTalk Application

The DuplexMEP application is composed of 2 artifacts, the SyncMagic8Ball orchestration and the DuplexMEP.Sync.WCF-NetTcp.ReceiveLocation. As mentioned earlier, the orchestration is a BizTalk version of the notorious Magic 8 Ball toy: it receives a request message containing a question, waits for a configurable number of seconds and then it randomly returns one of 20 standardized answers.

The following picture shows the receive location within the BizTalk Administration Console.

The following picture shows the structure of the SyncMagic8Ball orchestration.

If you are interested in more details about the DuplexMEP BizTalk application, you can read the first part of this article.

AppFabric Configuration

This section contains the steps I followed to configure the WCF workflow service in the IIS\AppFabric environment.

Using the IIS Manager Console, I created a web application called Magic8BallWF that points to the folder containing the project for my SyncMagic8Ball WCF workflow service.
Then I right-clicked the Magic8BallWF application, as shown in the picture below, I selected Manage WCF and WF Services from the context menu and then Configure.

On the Monitoring tab of the Configure WCF and WF for Application dialog, I enabled event collection to the AppFabric Monitoring database, I selected the connection string for the monitoring database and I chose an appropriate monitoring level.

On the Workflow Persistence tab, I selected the SQL Server Workflow Persistence option to enable persisting service instances to the AppFabric Persistence database.

Note
For data protection and performance reasons, WCF and WF services running on the same AppFabric environment can be configured to use separate monitoring and persistence stores. This solution is particularly suitable for a multi-tenant hosted environment running several applications that are managed by different companies or different divisions within the same company. For more information on this topic, you can read the following articles:

“Windows Server AppFabric Monitoring – Tracking Bottlenecks and Mitigation Techniques” post on the AppFabric CAT blog.
“Windows Server AppFabric How-To: Adding multiple persistence stores to an AppFabric installation” post on the AppFabric CAT blog.
“AppFabric Architecture and Deployment Topologies guide” on the Microsoft Download Center.

On the Workflow Host Management tab, I configured the runtime behavior of WF service instances.

Note
To manage a durable workflow service within the AppFabric runtime environment, the net.pipe binding must be configured for the website containing that application, and the net.pipe protocol must be enabled for the application. This is required because the Workflow Management Service (WMS), which works with the workflow persistence store to provide reliability and instance control, communicates with the Workflow Control standard endpoint of workflow services via the net.pipe protocol. If the net.pipe protocol is not set for a durable workflow application, when you attempt to configure the application, you will receive the following error message: “Workflow persistence is not fully functional because the net.pipe protocol is missing from the application’s list of enabled protocols.” To enable the net.pipe protocol for an application, right-click the application, point to Manage Application, and then click Advanced Settings. Add “,net.pipe” to “http” in the Enabled Protocols line (with no space between “http” and the comma), and then click OK. For more information on this topic, see "AppFabric Configuration Issues: .NET 4, net.pipe, and Role Services" on TechNet.

On the Auto-Start tab, I enabled the auto-start feature for my web application. When auto-start is enabled, hosted WF or WCF services within an application are instantiated automatically when the IIS service is started by the operating system. The services within the application will automatically start when the server is started. You can configure all services within an application to start, or a subset of services within an application. If you enable auto-start for an application, the auto-start feature will work only if you also enable auto-start for the application pool used by the application. The auto-start feature of AppFabric is built on top of the auto-start feature of Internet Information Services (IIS) 7.5, which is included in Windows 7 and Windows Server 2008 R2. In IIS, you can configure an application pool and all or some of its applications to automatically start when the IIS service starts. The AppFabric auto-start feature extends this functionality so that you can configure all or some of the services within an application to automatically start when the application starts.

For more information on this topic, see the following articles:
- "Auto-Start Feature" topic in the Windows Server AppFabric documentation.
- “Configure Auto-Start Using IIS Manager” topic in the Windows Server AppFabric documentation.
- "Configure Auto-Start Using Windows Server AppFabric Cmdlets" topic in the Windows Server AppFabric documentation.

On the Throttling tab, I configured the properties exposed by the ServiceThrottlingBehavior used by my WCF workflow service:
- MaxConcurrentCalls: specifies the maximum number of messages actively processing across a ServiceHost.
- MaxConcurrentInstances: specifies the maximum number of InstanceContext objects in the service that can execute at one time.
- MaxConcurrentSessions: specifies the maximum number of sessions a ServiceHost object can accept at one time.

Note
The default values for the properties MaxConcurrentCalls, MaxConcurrentInstances, MaxConcurrentSessions exposed by the ServiceThrottlingBehavior have been increased and made more dynamic as they are based on the number of processors seen by Windows.

Property	.NET 4.0 Default	Previous Default
MaxConcurrentCalls	16 * ProcessorCount	16
MaxConcurrentInstances	116 * ProcessorCount	26
MaxConcurrentSessions	100 * ProcessorCount	10

For more information on this topic, see “Less tweaking of your WCF 4.0 apps for high throughput workloads” post on the AppFabric CAT blog.

Finally I pressed the OK button to confirm settings.

Then, I opened the Services page, I right-clicked the SyncMagic8Ball and I selected Configure from the context menu. On the Monitoring tab, I clicked the Configure button on the right panel and selected the Troubleshooting Tracking Profile. This tracking profile is quite verbose and therefore is particularly helpful when debugging a WF service in testing environment, but it is not recommended in a production, unless you have to investigate and troubleshoot a problem.

The following table contains the configuration file of the WCF workflow service after completing these steps.

<?xml version="1.0" encoding="utf-8"?>
<configuration>
  <system.web>
    <compilation debug="true" targetFramework="4.0" />
  </system.web>
  <system.serviceModel>
    <bindings>
      <netTcpBinding>
        <binding name="netTcpBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpBinding>
      <netTcpContextBinding>
        <binding name="netTcpContextBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpContextBinding>
    </bindings>
    <client>
      <!-- This client endpoint is used by the WCF workflow service
           to incoke the WCF receive location exposed by the BizTalk -->
      <endpoint address="net.tcp://localhost:7171/Magic8BallBizTalk/Sync"
                binding="netTcpBinding"
                bindingConfiguration="netTcpBinding"
                contract="Magic8Ball"
                name="bizTalkSyncNetTcpBinding"/>
    </client>
    <services>
      <service name="SyncMagic8Ball">
        <endpoint address="" 
                  binding="basicHttpContextBinding" 
                  contract="IMagic8BallWF" 
                  name="basicHttpBinding_SyncMagic8Ball" />
        <endpoint address="" 
                  binding="netTcpContextBinding" 
                  bindingConfiguration="netTcpContextBinding" 
                  contract="IMagic8BallWF" 
                  name="netTcpBinding_SyncMagic8Ball" />
      </service>
    </services>
    <behaviors>
      <serviceBehaviors>
        <behavior>
          <!-- To avoid disclosing metadata information, set 
               the value below to false and remove the metadata 
               endpoint above before deployment -->
          <serviceMetadata httpGetEnabled="true" />
          <!-- To receive exception details in faults for debugging purposes, 
               set the value below to true. Set to false before deployment to 
               avoid disclosing exception information -->
          <serviceDebug includeExceptionDetailInFaults="true" />
          <!-- Added by AppFabric Admin Console -->
          <workflowInstanceManagement authorizedWindowsGroup="AS_Administrators" />
          <workflowUnhandledException action="AbandonAndSuspend" />
          <workflowIdle timeToPersist="00:01:00" timeToUnload="00:01:00" />
          <serviceThrottling maxConcurrentCalls="200" maxConcurrentSessions="200" maxConcurrentInstances="200" />
          <etwTracking profileName="Troubleshooting Tracking Profile" />
        </behavior>
      </serviceBehaviors>
    </behaviors>
    <serviceHostingEnvironment multipleSiteBindingsEnabled="true" />
  </system.serviceModel>
  <system.webServer>
    <modules runAllManagedModulesForAllRequests="true" />
  </system.webServer>
</configuration>

It’s probably worth noting that you can manually modifies the configuration file without using the administration extensions provided by AppFabric. However, AppFabric offers a convenient and handy way to accomplish this task.

Testing the Application

To test the application, you can proceed as follows:

Makes sure to start the DuplexMEP BizTalk application.
Open a new instance of the Client Application, as indicated in the picture below.
Enter an existential question like "Why am I here?", "What's the meaning of like?" or "Will the world end in 2012?" in the Question textbox.
Select one of NetTcpEndpointSyncWF in the Endpoint drop down list.
Specify a Delay in seconds in the corresponding textbox.
Press the Ask button.

Now, if you press the Ask button multiple times in a row, you can easily notice that the client application is called back by the WCF workflow service in an asynchronous way. Therefore, the client application doesn't need to wait for the response to the previous question before posing a new request.

Make some calls and then open the AppFabric Dashboard. This page is composed of three detailed metrics sections: three detailed metrics sections: Persisted WF Instances, WCF Call History, and WF Instance History. These sections display monitoring and tracking metrics for instances of .NET Framework 4 WCF and WF services. Let’s focus our attention on the WF Instance History section, highlighted in red in the figure below. The latter displays historical statistics derived from tracked workflow instance events stored in one or more Monitoring databases. It can draw data from several monitoring databases, if the server or farm uses more than one monitoring database for services deployed at the selected scope.

If you click the Completions link you can review WF instances that completed in the selected period of time. You can use the Query control on the Tracked WF Instances Page to run a simple query and restrict the number of rows displayed in the grid below.

Finally, you can right-click one of the completed WF instances and select View Tracked Events to access the Tracked Events Page where you can examine events generated by WCF and WF services. Here, you can group events by Event Type, as shown in the figure below, and analyze the user-defined events emitted by the current WCF instance using the CustomTrackingActivity that we saw at the beginning of this article.

In particular, you can quickly investigate the details of a selected event in the Details pane, as highlighted in red in the figure above.

Conclusions

In this article we have seen how to exchange messages with a WCF workflow service running in IIS\AppFabric using the Durable Duplex Correlation and a context binding. We have also seen how to create a custom activity to emit user-defined events and how to to use the using the AppFabric Dashboard to monitor custom tracking events generated by WF services. Finally we have seen how the Mapper activity provided by AppFabric Connect to implement message transformations in a WCF workflow service. This component not only allows to implement message transformations in a easy way in any WF project, but it allows developers to reuse maps from existing BizTalk application in an AppFabric solution. In the final part of this article, we’ll see how to implement an asynchronous communication between a WCF Workflow Service and an Orchestration using WS-Addressing and Content-Based Correlation. In the meantime, here you can download the companion code for this article. As always, your feedbacks are more than welcome!