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

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!

AD FS 2.0 to secure WCF and Workflow Services hosted in Windows Server AppFabric

SurenM — Wed, 15 Dec 2010 04:52:00 GMT

This blog entry is a sequel to the October 30th Blog “Securing WCF Services hosted in Windows Server AppFabric with Windows Identity Foundation (WIF)” where we demonstrated how to use WIF to secure Services hosted in AppFabric. And in this blog post we extend the same scenario to demonstrate how you can leverage Active Directory Federation Services 2.0 (AD FS 2.0) to provide you a solution for your ‘federated identity’ needs. The AD FS approach also provides you the solution that mitigates security risks (by using ActAs delegation) and authorizes access through all three tiers: the client, the middle tier (e.g., AppFabric hosted Services) to the data tier (e.g., databases) – we will cover this too.

This blog is targeted towards both the architect and developer community. The blog demonstrates how you can apply AD FS 2.0 into your architecture to provide federated identity solution and provides adequate details for you to attempt it in your project.

AD FS 2.0

AD FS 2.0 is a Server Role within Windows Server that provides support for WS-Federation, WS-Trust and SAML, with an easy to use management UI and a powerful claims processing rules engine. In a nutshell, it allows you to register your applications (Relying Parties), Claims Providers (e.g., other STS’s) and Attribute Stores (Identity Providers like Active Directory), and define the rules that govern how incoming claims are mapped to claims your application expects. The Figure 1 below illustrates these major features.

Figure 1: Major Features of AD FS 2.0 (Source MSDN.com)

One the biggest advantages you’ll notice almost immediately when getting started with AD FS is that you now have that “missing UI” for Security Token Service (STS) configuration that you don’t get when using the STS templates provided by Visual Studio.

Build Out

You can download AD FS 2.0 from here. In order to make use of AD FS 2.0 for our scenario, you will also need the following (see the Additional Resources for links):

· Active Directory

· WIF – Windows Identity Foundation and WIF .NET 4 SDK.

· Visual Studio 2010

To set up a test lab environment for this, the easiest approach is to have two VPC’s. One that only runs your Active Directory and the other that is joined to the former’s domain, and has AD FS 2.0, WIF, and VS2010 installed.

Lab Setup

In authoring this blog entry, we used a lab environment configured with two VPC’s. Step-by-step guidance on setting up the AD FS 2.0 test lab is available in the ‘references’ – the last section on this blog. Both these VPCs use a base Windows Server 2008 with Service Pack 2; details around these VPCs along with some major ‘steps’ are below:

VPC 1: FSAD – Domain Controller
- Add Domain Controller role: This will installs DNS
- Automatically and assign this machine a static IP address.
VPC2: FSWEB – AD FS Server & Development environmen
- Install .NET 3.5 SP1
- Assign this machine a static IP address
- Join domain; install AD FS 2 in federation server mode – this will install IIS
- Install VS 2010 (installs .NET 4).
- Create a self-signed SSL certificate.
- Register HTTPS binding on Default Web Site to use the self-signed cert.
- Complete AD FS Config Wizard selecting Stand-alone server option.Note: Before running on Server 2008 (not needed for R2) you need to add the entry below (or the service will fail to start) to Microsoft.IdentityServer.Servicehost.exe.config file, located in the ADFS 2 install directory (default C:\Program Files\Active Directory Federation Services 2.0).
  <runtime>
  <generatePublisherEvidence enabled="false"/>
  </runtime>
- Install WIF SDK (run as admin, be patient before clicking next on license screen, if it says please wait while computing disk space, restart computer and try again, waiting slightly longer).

Scenario

This blog builds upon the scenario presented in the previous blog (Securing WCF Services hosted in Windows Server AppFabric with Windows Identity Foundation (WIF)) which used a custom STS solution; this current blog will use AD FS 2.0 as the enterprise-grade STS. Figure 2 below provides a quick comparison of the scenarios in both the blogs –you will notice the big difference is previously we used a custom STS and this time around we used AD FS 2.0. With the release of AD FS2.0, we expect larger number of deployments taking a dependency on AD FS 2.0.

The coolest part about swapping in AD FS 2.0 from the previous scenario is that there are no have no code changes to make – all that was needed was a bit of configuration and this is what we will demonstrate in this blog.

Figure 2 a and b: Contrast scenarios using Custom STS (top) with current scenario of using AD FS 2.0 (bottom).

For this blog scenario we will update our architecture so that it leverages AD FS 2.0 as the STS, so that you are able to learn how to abstract usage of Identity Stores (referred to as Attribute Stores in AD FS interfaces) behind the STS using AD FS 2.0. Identity Stores could be Active Directory, MS SQL, LDAP, or any other custom provider. Subsequently in this blog posting we will also enhance the scenario presented to demonstrate integration of Microsoft SQL Server with Active Directory to append other identity related information to the claims.

The blog will also show to configure AD FS using its management MMC UI to configure the claims issuance rules for our applications, as well enabling ActAs delegation.

Implementation

In showing how to implement the scenario, we will start with the finished version of the Visual Studio solution we described in the previous blog posting and show how we can enhance it to leverage federated security, delegation and AD FS 2.0. To accomplish this we need will setup RulesEngine to use AD FS and configure AD FS to secure it; repeat this for the ComputationWorkflows project; and update the service references and related configuration in the RulesEngine and the DataWizApp. Figure 3, below shows our new implementation approach.

Figure 3: Using AD FS 2.0 STS to secure Services

Securing the RulesEngine with AD FS

To enable RulesEngine to use AD FS instead of the development STS, we run FedUtil, via Add STS Reference again. The key difference this time is the address we use for the federation metadata is on the Security Token Service screen we use the address of AD FS instead of the development STS. The following screens show the entire process.

Figure 4 to 9: Configuration Changes

The above configuration changes will take care of updating the RulesEngine web.config so that it calls AD FS during authentication.

Modifications to Web.Config files

We have two more modifications to make in web.config. We need to configure service clients (like DataWiz and RulesEngine) to request the desired claims; these claims are expressed in web.config. AD FS will only provide requested tokens, unlike our Dev STS which doesn't care and always sends the same set of tokens. In our scenario, FedUtil by default requests Name and Role, but we must manually request CanLoadData. This amounts to editing the custom binding added by FedUtil, and adding that claim to system.serviceModel\bindings\customBinding\binding\security\secureConversationBootstrap\issuedTokenParameters\additionalRequestParameters\SecondaryParamaters\Claims as shown below:

<trust:SecondaryParameters xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">

<trust:KeyType xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://docs.oasis-open.org/ws-sx/ws-trust/200512/SymmetricKey</trust:KeyType>

<trust:KeySize xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">256</trust:KeySize>

<trust:Claims Dialect="http://schemas.xmlsoap.org/ws/2005/05/identity"

xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">

<wsid:ClaimType Uri="http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"

Optional="true" xmlns:wsid="http://schemas.xmlsoap.org/ws/2005/05/identity" />

<wsid:ClaimType Uri="http://schemas.microsoft.com/ws/2008/06/identity/claims/role"

Optional="true" xmlns:wsid="http://schemas.xmlsoap.org/ws/2005/05/identity" />

<wsid:ClaimType

Uri="http://contoso.com/claims/canloaddata"

Optional="true"

xmlns:wsid="http://schemas.xmlsoap.org/ws/2005/05/identity" />

</trust:Claims>

<trust:KeyWrapAlgorithm xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://www.w3.org/2001/04/xmlenc#rsa-oaep-mgf1p</trust:KeyWrapAlgorithm>

<trust:EncryptWith xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://www.w3.org/2001/04/xmlenc#aes256-cbc</trust:EncryptWith>

<trust:SignWith xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://www.w3.org/2000/09/xmldsig#hmac-sha1</trust:SignWith>

<trust:CanonicalizationAlgorithm xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://www.w3.org/2001/10/xml-exc-c14n#</trust:CanonicalizationAlgorithm>

<trust:EncryptionAlgorithm xmlns:trust="http://docs.oasis-open.org/ws-sx/ws-trust/200512">http://www.w3.org/2001/04/xmlenc#aes256-cbc</trust:EncryptionAlgorithm>

</trust:SecondaryParameters>

</additionalRequestParameters>

</issuedTokenParameters>

In addition, AD FS exposes multiple service endpoints and when FedUtil runs, it basically chooses the first endpoint which is a certificate based one. Since we are using Windows domain credentials, we need to use a Kerberos endpoint. These endpoints are all listed in the web.config commented out under an alternativeIssuedTokenParameters section, you just need to remove the one that’s within the issuedTokenParameters section and replace it with one of the others from the commented region. The second highlighted line in the config above shows the updated endpoint in place.

Configuring AD FS for Securing the Solution

Now that we’ve configured the RulesEngine to use ADFS, we need to tell ADFS about the RulesEngine. To do this we open the AD FS 2.0 Management Snap-in from Start Menu->Administrative Tools->AD FS 2.0 Management. In the tree view we expand the Trust Relationships node and then right-click the Relying Party Trusts node, selecting Add Relying Party Trust… which loads wizard of the same name. We click Start past the welcome screen:

On the Select Data Source screen, we can simply import the Federation Metadata requirements of the RulesEngine from a file (since we are not using HTTPS to host the RulesEngine, the first radio is not an option). As you can see in the screenshot, the FederationMetadata.xml documentforRulesEngine is located under the RulesEngine project.

Next you enter a user friendly display name and any notes.

Then, by default we don’t want to perform any authorization at the ADFS STS, so we just leave the Permit all option selected.

You can review the various settings pulled in from the FederationMetadata and your selections on the following screen.

You will then be prompted with the Edit Claims Rules screen, which you’ll want to display to configure how claims are issued from ADFS for the RulesEngine. In our scenario, we need to define rules to issue three claims: Name, Role and CanLoadData. We will also define a fourth, DOB, to show how we can obtain values from claims from attribute stores other than Active Directory. The following screenshot shows our Issuance Transform Rules with all claims added; we will then see how we configure the Name, Role and CanLoadData claims (the first three in the screenshot). We’ll return to showing how we add the DOB claim at the end of this blog post.

Let’s start with the Name claim. Here we receive the name via the presented Windows credentials, so we just need to pass it through. To create such a rule, click Add Rule… and the in the Claim rule template select Pass Through or Filter and Incoming Claim.

Click Next. Enter a user friendly description for the rule. Because we are after the incoming Name claim, select that in the first drop down and leave Pass through all claim values selected. Your screen should look as follows:

Clicking Finish will add the rule to the Issuance Transform Rules listing, but more importantly results in the Name claim being properly forwarded from the AppWiz client to the RulesEngine.

Next, let’s see how we add the Role claim that always provides a value of “AuthenticatedUsers”. Again, we click Add Rule… here we simply want to issue a new Role claim (we don’t want to pass through or transform the existing one). The only way to issue a completely new claim is to use a Custom Rule. So in the Claim rule template, select Send Claims Using a Custom Rule and click next.

All rules that you configure through the rule templates are ultimately expressed in this rule syntax (see the Additional References section for details on the syntax).

Note: A quick point, for any rule you’ve configured that’s not a Custom Rule, you can edit it and click View Rule Language… to see it- this is a great way to bootstrap your custom rule authoring.

Here we write a custom rule, that basically reads irrespective of any incoming claim (the condition that would normally be to the left of the => is empty means this), create and output to the claimset a new claim with the “role” type and a value of “AuthenticatedUsers”. Clicking Finish adds this rule and now we have our second claim being issued.

The third claim we add is for CanLoadData. Recall users who belong to the SeniorManagers Windows group should get this claim with a value of true. To add this rule, again we begin by clicking Add Rule… this time selecting Send Group Membership as a Claim as and clicking Next.

Now we give the rule a friendly name, select the SeniorManager group from AD, specify that the outgoing claim type is canloaddata and give it a value of true as follows:

After clicking Finish, we have our third rule in place. Now SeniorManagers will automatically get the canloaddata claim as expected by our scenario.

Back on the Edit Claims Rules for RulesEngine screen, if you click the Issuance Authorization Rules tab you should see this single rule called Permit Access to All Users. This is what we want, as we don’t want to perform any authorization at the STS in our scenario (we just want it to authenticate). We could add similar rules here that control if a user is even granted tokens to present to the RulesEngine.

Securing the ComputationWorkflows with AD FS

Securing the ComputationWorkflows project happens in a way similar to what we did for the RulesEngine. We add an STS Reference to AD FS from the ComputationWorkflows project, this time using the address for the LoadService.xamlx as the value of the Application URI in the first screen of FedUtil. Incidentally, this actually ends up taking care of both services because the service endpoints are not explicitly described in the web.config as it uses a protocolMapping. We run through the rest of FedUtil following exactly the same steps as we did for RulesEngine.

Next, we need to register the ComputationWorkflows application with ADFS. We add the Relying Party Trust as previously described, but this time use the FederationMetada.xml located under the ComputationWorkflows project. We then create the following issuance rules:

The first three are exactly the same as those we added to RulesEngine. The fourth one is new. This one is responsible for taking the CanLoadData claim acquired and passed down by the RulesEngine and passing it thru to the ComputationWorkflows when it performs the ActAs delegated call. This rule uses the Pass Through or Filter an Incoming Claim rule template and is configured as follows:

Just as for the RulesEngine, we don’t specify any authorization at the STS, so our Issuance Authorization Rules tab contains just the single Permit Access to All Users entry.

Now, the ComputationWorkflow has a unique requirement because it is the target of invocations that use delegated, ActAs credentials. By default, delegation is disabled. In order to allow it, on the Delegation Authorization Rules tab, we click Add Rule… and select Permit All Users in the drop down and click Next.

Clicking Finish will allow any callers to use delegated credentials. In the real world, you will likely choose a specific account that is allowed to perform delegation (such as NETWORK SERVICE) and then specify a Permit or Deny Users Based on an Incoming Claim rule template that checks for that account name.

With that we have fully configured AD FS for both services, now we just need to update the service references so the clients know to authenticate against AD FS.

Update the Service References in the RulesEngine

First, in Visual Studio, from the RulesEngine project update the service reference to the Load service so that it gets the new federation settings. We need to adjust web.config so that the Download endpoint uses the same settings as were automatically applied to Load when we updated the service reference. Open web.config and update the client endpoint for the Download service. Change the binding to custom and the bindingConfiguration value so it uses the same configuration name as Load. Here’s an example:

<system.serviceModel>

<endpoint address="http://localhost/ComputationWorkflows/DownloadService.xamlx"

binding="customBinding" bindingConfiguration="WS2007FederationHttpBinding_IService2"

contract="Download.IService" name="WS2007FederationHttpBinding_IService">

</identity>

</endpoint>

<endpoint address="http://localhost/ComputationWorkflows/LoadService.xamlx"

binding="customBinding" bindingConfiguration="WS2007FederationHttpBinding_IService2"

contract="Load.IService" name="WS2007FederationHttpBinding_IService1">

</identity>

</endpoint>

</client>

Update the DataWizApp client's Service Reference

Simply update service reference on Contoso (rules engine) and the app.config is updated for federation with AD FS.

Because this client will need to also request CanLoadData, add the canloaddata claim to the requested claims and select the Kerberos ADFS endpoint(by modifying the SecondaryParameters Claims collection in the custom binding as we had shown previously when we configured the RulesEngine to use AD FS).

Integrating AD FS with other data stores: Provide additional claim information in the form of attributes acquired from Microsoft SQL Server. Extra Credit J

While Active Directory might store organization and contact information, it would not likely store other user information such as preferences, favorites, authorized working hours or similar items useful to the application. Here we show how we can leverage multiple attribute stores simultaneously (e.g., Active Directory & SQL) from ADFS to provide richer claims about the identity.

For simplicity we will use the following database create in our local copy of SQL Express. It contains a single table called Users, which itself has three columns: Id, Name and DOB:

/****** Object: Table [dbo].[Users] Script Date: 12/11/2010 14:27:05 ******/

IF EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[dbo].[Users]') AND type in (N'U'))

DROP TABLE [dbo].[Users]

/****** Object: Table [dbo].[Users] Script Date: 12/11/2010 14:27:05 ******/

SET ANSI_NULLS ON

SET QUOTED_IDENTIFIER ON

IF NOT EXISTS (SELECT * FROM sys.objects WHERE object_id = OBJECT_ID(N'[dbo].[Users]') AND type in (N'U'))

BEGIN

CREATE TABLE [dbo].[Users](

[Id] [int] IDENTITY(1,1) NOT NULL,

[Name] [varchar](255) COLLATE SQL_Latin1_General_CP1_CI_AS NOT NULL,

[DOB] [date] NOT NULL

)

END

SET IDENTITY_INSERT [dbo].[Users] ON

INSERT [dbo].[Users] ([Id], [Name], [DOB]) VALUES (3, N'CONTOSO\John Doe', CAST(0xDF080B00 AS Date))

INSERT [dbo].[Users] ([Id], [Name], [DOB]) VALUES (4, N'CONTOSO\Jane Doe', CAST(0x3F080B00 AS Date))

SET IDENTITY_INSERT [dbo].[Users] OFF

You may want to add additional records that match the names of users you will test with.

After creating the database, you need to create a login for Network Service (or the account under which AD FS is running) and give it database owner permissions (a little overkill, but for simplicities sake).

Now we can look at what it takes to configure AD FS to query the Users table. First we need to register the database by adding a SQL Attribute Store. Within the AD FS 2.0 MMC, right-click the Attribute Stores folder and select Add Attribute Store… in the dialog that appear provide a user friendly display name, select SQL as the Attribute store type and enter the connection string and click OK:

Now we can access this DB for querying from our claims issuance rules, so let’s add a rule that adds the DOB value queried for the user from the Users table to the set of emitted claims. For RulesEngine, add custom rule that looks up DateOfBirth DOB. To do this, go to the Relying Party Trusts folder. Right-click on the RulesEngine entry and select Edit claims rules… on the Issuance Transform Rules tab click Add Rule… IN the first dialog, select the Send Claims Using a Custom Rule template (queries to other Attribute Stores always use custom rules) and click Next. Define the rule as follows:

Notice the syntax for the query parameter in the issue statement is parameterized to use the value of the incoming Name claim in an approach similar to String.Format(). Note that the parameter {0} is not surrounded by single quotes as you might be inclined to do…adding these quotes will generate an error.

There are a few items to point out when writing a custom rule for this. First, in the select statement a column maps to a claim type (where the types parameter in the rule lists out the types in order). Returning multiple rows results in multiple claim values for that claim type. In our case, we have only one column and one row, so we emit exactly one claim of the DOB type with the value of the DOB field in the database. The second gotcha, is that all values must of a string form (e.g., varchar, char, text, etc) which is why in the above we needed to cast DOB because in the database is stored as a Date. Finally, the value set for the store parameter is the Display Name you entered when you registered the new Attribute Store.

With this rule in place, you can access the DOB claim in service code as you would any other (e.g., Service.svc.cs), even within a workflow service (following the steps we showed in the first blog entry)!

Conclusion

In conclusion, you saw how it was possible to replace a custom STS solution with AD FS 2.0 that that was needed to do so was configuration changes. And more importantly we also demonstrated how easy it is to integrate data from other stores to provide other claim data/attributes. I do hope this blog post series helped you with your architectural decisions around securing Services hosted by Windows Server AppFabric.

Additional Resources

Prequel Blog - Securing WCF Services hosted in Windows Server AppFabric with Windows Identity Foundation (WIF): http://blogs.msdn.com/b/appfabriccat/archive/2010/10/30/securing-wcf-services-hosted-in-windows-server-appfabric-with-windows-identity-foundation-wif.aspx
ADFS 2.0: http://www.microsoft.com/downloads/en/details.aspx?FamilyID=118c3588-9070-426a-b655-6cec0a92c10b&displaylang=en
WIF Runtime: http://www.microsoft.com/downloads/en/details.aspx?FamilyID=eb9c345f-e830-40b8-a5fe-ae7a864c4d76&displaylang=en
WIF SDK: http://www.microsoft.com/downloads/en/details.aspx?FamilyID=c148b2df-c7af-46bb-9162-2c9422208504&displaylang=en
Setting up an AD FS 2.0 with WIF lab environment: http://technet.microsoft.com/en-us/library/adfs2-federation-wif-application-step-by-step-guide(WS.10).aspx
AD FS Claims Rules Language Syntax: http://technet.microsoft.com/en-us/library/dd807118(WS.10).aspx

Acknowledgement

Acknowledge contributions from Mark Simms and Paolo Salvatori

Namaste!

Logging and Counters in App Fabric Cache

Jaime Alva — Wed, 15 Dec 2010 01:41:00 GMT

The following is a good place to get started with how to go about collecting basic information for monitoring and/or troubleshooting an App Fabric (AF) cache setup (Host/Cluster and clients). It starts by pointing to some resources on the different available logging features and then, via a sample scenario, goes over the decision taken to implement a logging solution. Additionally, it answers a frequent customer question on what are the basic recommended performance counters to collect.

What App Fabric Cache offers

Here is a quick walk through on how to better format and generate the log generation for ease of management, some of the links within are pre-release so you may refer to the following more updated ones, server log sink settings and client log sink settings. These should give a fairly good idea on the logging capabilities offered in AF Cache, please review the given links, as its knowledge will help further reading. With these concepts, the discussion and planning on what is the best suitable logging solution for your specific implementation can start.

A Sample Scenario

Assuming that memory pressure issues are a concern on the host side. The default event trace level of ERROR would not be enough as it would be necessary to have a more detailed sense of what objects are being cached on the host. This can be done by overriding the default host log sink to collect information level logs, enabling more detailed log analysis in the case of memory related errors. 5 different levels are provided: No Tracing (-1), Error (0), Warning (1), Information(2), Verbose(3). In this sample, the Information level will be taken.

At that point the next decision will be to determine if the configuration setting should be performed via code or XML (configuration file). In this sample, the organization decides that their Infrastructure personnel can handle the required changes via XML and no programmers will be required (no code needed) and hence the XML solution is the simplest.

Next is the type of logging – as the same infrastructure team will also be analyzing the logs, a file-based log sink is agreed upon (versus console or Event Tracing for Windows - ETW). For the sake of simplicity and to ease the understanding of the sample in this blog, AF Cache logging was chosen. ETW logging is a viable option as well. Since the logs will be written into an existing central shared location on the network, the NETWORK SERVICE account is given rights to the share (in the case of a cluster, each host NETWORK SERVICE account will have to be added to have write access to this share). NOTE that AF Cache service ONLY runs under the NETWORK SERVICE account of the server (which is the account assigned to the computer by the domain), the service cannot run as any other account, such as an independent (not server/local machine account) Network user or group.

In the case of an outage the logs could be overwritten, to alleviate this, the process-specific character ($) is agreed upon and it is to be used within the log name. Also, the log generation interval is settled for every hour (dd-hh).

Similarly, since memory pressure on the webservers (AF cache client) is also a concern, the client logs sink needs similar changes. The final custom type attribute for client and host for the fabric object will then look similar to the following:

<customType

className="System.Data.Fabric.Common.EventLogger,FabricCommon"

sinkName="System.Data.Fabric.Common.FileEventSink,FabricCommon"

sinkParam=\\CentralLogs\\AFCache\\Server1-$/dd-hh

<!-- For the client machines the log name are modified: sinkParam="\\CentralLogs\\AFCache\\Client1-$/dd-hh" –>

defaultLevel="2"

/>

Logs are a good way to collect application specific or, as in the case above, scenario specific information that will allow for ad-hoc or error-driven analysis. Similarly, collecting performance counters can give a window in the internal operations of not only the particular application (AF cache) but also of the overall system.

Performance Counters

As a complement to the above, here is a list of recommended performance counters to aid in monitoring (for later analysis) and troubleshooting AF cache issues. The table contains the names for the performance monitor counters and the important instances to monitor. A small comment follows each counter grouping.

Counter Name	Running Instance	Counter Instance	Comments
AppFabric Caching:Host	N/A	All (*): Cache Miss Percentage,Total Client Requests, Total Client Requests /sec, Total Data Size Bytes, Total Evicted Objects, Total Eviction Runs, Total Expired Objects, Total Get Requests, Total Get Requests /sec, Total GetAndLock Requests, Total GetAndLock Requests /sec, Total Memory Evicted, Total Notification Delivered, Total Object Count, Total Read Requests, Total Read Requests /sec, Total Write Operations, Total Write Operations /sec	For obvious reasons all the host counters are included. For memory troubleshooting and monitoring, the counters “Total Data Size Bytes” and “Total Object Count” are relevant. Also, a high level of evictions (Total Memory Evicted) may indicate memory pressure.
.NET CLR Memory	Distributed CacheService	# Gen 0 Collections, # Gen 1 Collections, # Gen 2 Collections, # of Pinned Objects, % Time in GC, Large Object Heap size, Gen 0 heap size, Gen 1 heap size, Gen 2 heap size	These counters will give a fuller spectrum on what is taking place with the CLR memory. For instance, a large “% Time in GC” would indicate that too much garbage collection (GC) is taking place, and the CPU will provably be working extra to process the GC. Memory pressure may therefore be an issue. See this blog for further details
Memory	N/A	Available MBytes	The preference will be to keep this at above 10% of total memory and plan to have available space for high traffic times.
Process	Distributed CacheService	% Processor Time, Thread Count, Working Set	A small working set could imply memory pressure from AF Cache as oppose to other processes in the box.
Process	_Total	% Processor Time	Monitoring the total process time will tell which application(s) are competing for CPU resources.
Network Interface		All (*): Bytes Received/sec, Bytes Sent/sec, Current Bandwidth	As long as memory and Garbage Collection are not an issue then the CPU should be expected to work well and a lot of throughput can be handle. At this point a slow network card or any network point from there to the client may become the bottleneck. Monitor the network interfaces to ensure that they are not saturated.

Adding the counters

To avoid having to include all the counters above one-by-one, download this performance counter data collection set template and import it to the server, as follows (alternative you can also use this more generic instructions):

From the run command type “perfmon” and hit enter.
From the tree view, open “Data Collector Sets”, right click on “User Defined” and then choose “New” > “Data Collector Set”, this screen should appear, enter a suitable name and choose “Create from a template (Recommended)”.
Click “Next” and on do not choose any of the basic templates, instead click “Browse…”, load the downloaded XML file template
Click “Next” and Browse to the drop directory you wan to setup, make sure it is not already been use by other counters, otherwise it may not work or maybe to difficult to recognize one counter from another, and click “Finish”. Alternatively, click on “Next” and then choose “Open properties for this data collector set” and you will get a chance to review all the settings before finishing.
Right click on the recently created Data collector set and hit Start, as shown
Click “stop” once you are done collecting counters and collect the counters from the drop directory.

In summary

Both logs and performance counters collected together are the first step in being ready to analyze errors or monitor for specific concerns or conditions (i.e. memory pressure) with AppFabric Caching.

Since this is a big subject, I will look into further exploring the reasons behind the performance counters recommendation in a future blog.

Author: Jaime Alva Bravo

Reviewers: Mark Simms; James Podgorski; Rama Ramani

StreamInsight Query Patterns: Fun with process data – calculating the wind chill factor

Mark Simms [MSFT] — Tue, 14 Dec 2010 00:17:00 GMT

Process data, commonly accessed from OPC or a process historian such as OSIsoft PI, is well suited for analysis using StreamInsight – being time-series data with data values from a range of sensors. In this post, I’ll walk through a couple of simple patterns for working with process data (not that these patterns are isolated to process data).

To follow along, the LINQ query for this example is here (if you haven’t used LINQPad with StreamInsight yet, follow the directions here to get up and running).

The general case for collecting process data is that of a range of sensors (temperature, pressure, humidity, flow, etc) connected to process control equipment (Programmable Logic Controllers, or PLC’s, SCADA systems, etc) that in turn relay the information upstream to process historians (or provide a direct data gathering interface, through OPC or a similar protocol).

The reason I mention this level of detail has to do with how process data typically arrives – in a single or mixed stream. That is to say as connections to these embedded devices tends to be ‘expensive’ we want to minimize connections and communication and pull multiple data values over a single connection, often emitted as a single CepStream<T> in StreamInsight (rather than having, for example, temperature and humidity information coming from two different data sources, in two different streams). This is why I have intermingled data types in my examples below, and why the query patterns usually start out with creating different virtual streams based on a data type.

For these examples, let’s consider a representative process control data type with some simple fields

string Id
DateTime Timestamp
double Value

Next, assume that the Id field uses an “asset.sensor” naming convention (i.e. Station1.Temperature, and Station1.Windspeed). Also assume that the data updates are not guaranteed to be synchronized (i.e. the underlying PLC may not report updates to temperature and humidity at the same time, nor on the same schedule). For the first query, we want to create a query that answers the question:

“What is the current wind chill factor” (or is it a dry cold - the opposite side of the coin from the humidex)

Which breaks down to answering these questions:

For each station, what is the latest value of both temperature and wind speed?
Every time I have a new temperature or humidity value, I need to calculate a new wind chill factor.

Let’s start by examining the first question. Whenever we see a question involving latest value, we need to take a series of point or interval events and convert them into a signal (refer to Point to Signal conversion on MSDN).

This is a timeline representation of what we’re looking to do. We need to:

Create separate streams for the temperature values and wind speed values, creating the inferred join key based off of the prefix of the Id.
Convert each stream from a series of Point Events into signal stream using AlterEventLifetime and ClipEventDuration.
For each joint event (i.e. each time we receive a new value from either sub-stream, calculate a new wind chil).

For the purposes of this example, assume that all temperature are in F, and the wind speed is in mph.

/////////////////////////////////////////////////////////////////////////////
// Create some sample values for both temperature and wind sensor 
// measurements
/////////////////////////////////////////////////////////////////////////////
var sensorValues = new[]
{
    new { Id = "Station10.Temperature", Timestamp = DateTime.Parse("10/23/2009 9:00:00 AM"), Value = 20.1f },
    new { Id = "Station10.Temperature", Timestamp = DateTime.Parse("10/23/2009 9:00:30 AM"), Value = 18.1f },
    new { Id = "Station10.Temperature", Timestamp = DateTime.Parse("10/23/2009 9:00:45 AM"), Value = 17.1f },
    new { Id = "Station10.Temperature", Timestamp = DateTime.Parse("10/23/2009 9:01:00 AM"), Value = 22.5f },
    
    new { Id = "Station10.Windspeed", Timestamp = DateTime.Parse("10/23/2009 9:00:05 AM"), Value = 30.1f },
    new { Id = "Station10.Windspeed", Timestamp = DateTime.Parse("10/23/2009 9:00:25 AM"), Value = 40.4f },
    new { Id = "Station10.Windspeed", Timestamp = DateTime.Parse("10/23/2009 9:00:35 AM"), Value = 50.3f },
    new { Id = "Station10.Windspeed", Timestamp = DateTime.Parse("10/23/2009 9:01:55 AM"), Value = 40.6f },
};
 
/////////////////////////////////////////////////////////////////////////////
// Convert the input set into a temporal stream by using the Timestamp 
// field (and ordering the incoming data)
var inputStream = sensorValues.OrderBy(e => e.Timestamp)
    .ToPointStream(Application, e =>
        PointEvent.CreateInsert(e.Timestamp, e),
        AdvanceTimeSettings.StrictlyIncreasingStartTime);
 
// Create a temperature stream, and assign the ID as the component of the 
// ID before the .Temperature
var temperatureStream = from e in inputStream
                        where e.Id.EndsWith(".Temperature")
                        select new
                        {
                            Id = e.Id.Substring(0, e.Id.IndexOf(".")),
                            Timestamp = e.Timestamp,
                            Value = e.Value
                        };
 
// Create a windspeed stream, and assign the ID as the component of the 
// ID before the .Windspeed
var windspeedStream = from e in inputStream
                        where e.Id.EndsWith("Windspeed")
                        select new
                        {
                            Id = e.Id.Substring(0, e.Id.IndexOf(".")),
                            Timestamp = e.Timestamp,
                            Value = e.Value
                        };
 
// TEMP: dump the raw streams
//temperatureStream.Dump("temps");
//windspeedStream.Dump("wind");
 
// Convert the temperature and wind speed streams into signals (i.e. remember
// last known value).  Assume that we get at least one update every hour
var temperatureSignal = temperatureStream
    .AlterEventDuration(e => TimeSpan.FromHours(1))
    .ClipEventDuration(temperatureStream, (e1, e2) => (e1.Id == e2.Id));
 
var windspeedSignal = windspeedStream
    .AlterEventDuration(e => TimeSpan.FromHours(1))
    .ClipEventDuration(windspeedStream, (e1, e2) => (e1.Id == e2.Id));
 
// TEMP: dump out the signal streams
//var sinkTemp = from p in temperatureSignal.ToIntervalEnumerable()
//            where p.EventKind == EventKind.Insert
//            select new { p.Payload.Value, p.Payload.Id, p.StartTime, p.EndTime };
//sinkTemp.Dump("temp dump");
//
//var windTemp = from p in windspeedSignal.ToIntervalEnumerable()
//            where p.EventKind == EventKind.Insert
//            select new { p.Payload.Value, p.Payload.Id, p.StartTime, p.EndTime };
//windTemp.Dump("wind dump");
 
 
// Join the two streams based on ID - the output will be an event containing
// the last known values for both Temperature and Windspeed.
var measurements = from e1 in temperatureSignal
                    join e2 in windspeedSignal on e1.Id equals e2.Id
                    select new
                    {
                        Id = e1.Id,
                        Temperature = e1.Value,
                        Windspeed = e2.Value
                    };
 
// TEMP - dump out the joint values
var measures = from p in measurements.ToIntervalEnumerable()
                where p.EventKind == EventKind.Insert
                select new { p.Payload.Temperature, p.Payload.Windspeed, p.Payload.Id, p.StartTime, p.EndTime };
 
measures.Dump("measures");

This LINQ snippet implements step one and two, giving us the joined result of the last known values for each sub-stream:

Now that we have the correlated temperature and wind speed in a single event, we can go ahead and calculate the wind chill factor for each data update. The algorithm we’ll use, as defined by the National Weather Service is

with T being the e.Temperature value, and V being the wind speed factor. We add this to the projection as

// Join the two streams based on ID - the output will be an event containing
    // the last known values for both Temperature and Windspeed.
    var measurements = from e1 in temperatureSignal
                       join e2 in windspeedSignal on e1.Id equals e2.Id
                       select new
                       {
                            Id = e1.Id,
                            Temperature = e1.Value,
                            Windspeed = e2.Value,
                            Windchill = Math.Round( 35.74
                                + (0.6215 * e1.Value) 
                                - (35.75 * Math.Pow(e2.Value, 0.16)) 
                                + (0.4275 * e1.Value * Math.Pow(e2.Value, 0.16)))
                       };
    

And observe the calculated results.

This query pattern will work for any data source wherein you need to work off of and synchronize the “last known value” for a given set or subset of data. Note that if you are guaranteed to have synchronized timestamps on incoming data the point to signal conversion steps are not necessary before performing the join (as the timestamps of the incoming events already have the appropriate overlap).

SQL Azure and Entity Framework Connection Fault Handling

James Podgorski — Sat, 11 Dec 2010 01:39:00 GMT

A number of customers are choosing to use Entity Framework with Microsoft SQL Azure, and rightfully so. Entity Framework provides an abstraction above the database so that queries are developed using the business model or conceptual model and queries are executed using the richness of LINQ to Entities. SQL Azure is a cloud-based database based upon SQL Server which provides reliability and elasticity for database applications. And, the Entity Framework supports SQL Azure, both the designer and the provider runtime can be used against either a SQL Server database on premise or in the cloud.

But these same customers are describing connectivity drops, classified as general network issues, and are asking for the best practice to handle such network issues between SQL Azure instances and clients utilizing Entity Framework. In fact, SQL Azure has a whole host of reasons to terminate a connection including but not exclusive to resource shortages and other transient conditions. Similar issues apply when using ADO.NET; networks can be unreliable and are often suspect to sporadic fits resulting in dropped TCP connections. There are a number of blog posts, such as 'SQL Azure: Connection Management in SQL Azure' and 'Best Practices for Handling Transient Conditions in SQL Azure Client Applications', which provide connection fault handling guidance and even a framework for creating retry policies with SQL Azure client. Neither article is fully comprehensive regarding the Entity Framework aspects. The purpose of this blog posting is to fill in the details and discussion points for the many developer options to handle and recover from intermittent connectivity drops when using Entity Framework.

Background Information – Connection Pooling

It is well known that the creation and teardown of database connections to SQL Server is expensive, thus ADO.NET uses connection pools as an optimization strategy to limit the cost of the database connection creation and teardown operations. The connection pool maintains a group of all physical connections to the database and when clients such as Entity Framework request a connection, ADO.NET provides the next available connection from the pool. The opposite occurs when clients close a connection – the connection is put back into the pool. What is less understood is that the connection pooler will remove connections from the pool only after an idle period OR after the pooler detects that the connection with the server has been severed. But the pooler can only detect severed connections after an attempt has been made to issue a command against the server. This means that clients such as Entity Framework could potentially draw a severed or invalid connection from the pool. With high latency and volatile networks, this happens on a more frequent basis. The invalid connections are removed from the pool only after the connection is closed. Alternatively the client can flush all of the connections using ClearAllPools or ClearPool methods. See MSDN article ‘SQL Server Connection Pooling' for a more verbose description of the concepts.

Background Information - Entity Framework Database Connection

The Entity Framework provider abstracts most if not all of the facets from executing a query against the backend store, from establishment of the connection, to the retrieval of the data and materialization of the POCO or EntityObjects. Nevertheless it does provide access to the underlying store connection through the Connection property of the System.Data.Objects.ObjectContext.

The ObjectContext wraps the underlying System.Data.SqlClient.SqlConnection to the SQL Server database by using an instance of the System.Data.EntityClient.EntityConnection class. The EntityConnection class exposes a read/write StoreConnection property which is essentially the underlying SqlConnection to the SQL Azure instance. That mouthful simply says that we have a mechanism upon which to read the current state of a connection and assign the store connection if so desired. Since we have access to the database connection, we can most certainly catch exceptions thrown when the network transport has failed and retry our operations per a given retry policy.

One subtlety which requires a touch of clarification, take the code snippet below. In this example the connection was explicitly opened on the context. The explicit opening of the connection is a way of informing EF not to open and reopen the connection on each command. Had I not opened the connection in this way, EF would implicitly open and close a database connection for each query within the scope of the context. We will leverage this knowledge in the connection retry scenarios that follow.

using (AdventureWorksLTAZ2008R2Entities dc = new AdventureWorksLTAZ2008R2Entities())
{
    dc.Connection.Open();
    // ...
}

Case #1 – Retry Policies

Let’s take the following code as an example. How many connections are drawn from the pool? The answer is two, one for the retrieval of the customer and another for the retrieval of the address. This means that if implemented in this general way, Entity Framework will use a new connection for every query it executes against the backend data store given the scope of the current ObjectContext. It pulls a connection for the pool, submits the query against the database and closes the connection for each LINQ query. You can see this by running SQL Profiler on an instance of SQL Server as shown in the table below the query.

using (AdventureWorksLTAZ2008R2Entities dc = new AdventureWorksLTAZ2008R2Entities())
{
    int cId = 29485;
    Customer c1 = (from x in dc.Customers
                   where x.CustomerID == cId
                   select x).First();

    Address ad1 = (from x in dc.Addresses
                   from y in dc.CustomerAddresses
                   where y.CustomerID == cId && x.AddressID == y.AddressID
                   select x).FirstOrDefault();
}

Audit Login


RPC:Completed

	exec sp_executesql N'SELECT TOP (1) [Extent1].[CustomerID] AS [CustomerID] …
Audit Logout


RPC:Completed

	exec sp_reset_connection
Audit Login


RPC:Completed

	exec sp_executesql N'SELECT [Limit1].[AddressID] AS [AddressID], …
Audit Logout

But there is an interesting facet to the code above, the connection to the database was not explicitly created. The EntityConnection was transparently created by the Entity Framework when the AdventureWorksLTAZ2008R2Entities ObjectContext was instantiated and the StoreConnection property or actual SqlConnection state is set to closed. Only after the LINQ query is executed by the EF provider is the connection opened and the query submitted to SQL Server. The state of the inner connection changes to closed and the connection placed back into the pool once the results have been successfully retrieved.

In this example, we have two places in which a transient network error or invalid connection in the pool could affect the query and cause a System.Data.EntityException, for the retrieval of the customer and for the retrieval of the address. The inner exception of the EntityException is of type System.Data.SqlClient.SqlException and it contains the actual SQL Server error code for the exception.

A policy can be applied to wrap the LINQ queries to catch the EntityException and retry the query given the particular metrics of the policy. The retry policy in the code below utilizes the Transient Conditions Handling Framework described in the blog written by my teammate Valery. This blog provides a very comprehensive selection of retry policies which will properly handle the thrown exceptions. The basic principle is to support a number of retries with increasing periods of wait between each subsequent retry (i.e. a backoff algorithm). In this case we abort the operation after 10 attempts with a wait periods of 100ms, 200ms… up to 1 second.

using Microsoft.AppFabricCAT.Samples.Azure.TransientFaultHandling;
using Microsoft.AppFabricCAT.Samples.Azure.TransientFaultHandling.SqlAzure;

using (AdventureWorksLTAZ2008R2Entities dc = new AdventureWorksLTAZ2008R2Entities())
{
    int cId = 29485;
    int MaxRetries = 10;
    int DelayMS = 100;

    RetryPolicy policy = new RetryPolicy<SqlAzureTransientErrorDetectionStrategy>(MaxRetries, TimeSpan.FromMilliseconds(DelayMS));

    Customer c1 = policy.ExecuteAction<Customer>(() => 
                  (from x in dc.Customers
                   where x.CustomerID == cId
                   select x).First());

    Address ad1 = policy.ExecuteAction<Address>(() =>
                  (from x in dc.Addresses
                   from y in dc.CustomerAddresses
                   where y.CustomerID == cId && x.AddressID == y.AddressID
                   select x).FirstOrDefault());
}

This retry policy approach is not without its challenges, particularly in regards to the developer experience. The Func delegates require that a type be passed in, something that is somewhat cumbersome when using anonymous types because the return type must be set to an object type. This means that the developer must cast the object to the anonymous type to make use of it, a sample of which is shown below. I created a CastHelper class for that purpose.

public static class CastHelper
{
    public static T Cast<T>(object obj, T type)
    {
        return (T)obj;     
    }
}

var c1 = policy.ExecuteAction<object>(() =>
              (from x in dc.Customers
               where x.CustomerID == cId
               select new { x.CustomerID, x.FirstName, x.LastName }).First());

var anon = CastHelper.Cast(c1, new { CustomerID = -1, FirstName = "", LastName = "" });

Case #2 – Retry Policy With Transaction Scope

Case #2 expands the previous case to include a transaction scope. The dynamics of the sample changes because a System.Transactions.TransactionScope object is used to ensure data consistency of all scoped queries using an ambient transaction context which is automatically managed for the developer. But from the trace shown below we still observe the typical pattern of Audit Logout events with the SQL Transaction Begin and End events. This begs the question: How can we have a local transaction (non-distributed) if our data access code spans multiple connections? Well, in this case we are not truly spanning multiple physical connections because every connection drawn from pool is based upon the proper connection string and, for those enlisted in a transaction, the thread context. We can imagine the connection pool subdivided into subdivisions based upon individual transaction contexts and the connection string. In this way, pulling a connection from the pool guarantees that we are not attempting to enlist in a distributed transaction.

The net result: We cannot implement our retry policies at the query or SaveChanges level and still maintain the ACID properties of the transaction. The retry policies must be implemented against the entire transactional scope as shown below. Note that if you do attempt to place retry logic against the individual queries and a network glitch occurs, be assured than an EntityException will be thrown having an inner SqlException with a message of “MSDTC on Server “xxx” is unavailable”, SQL Azure does not support distributed transactions. Nevertheless, this is not SQL Azure specific problem. The error is non-recoverable at the query statement level; it is all or nothing with transactions.

using (AdventureWorksLTAZ2008R2Entities dc = new AdventureWorksLTAZ2008R2Entities())
{
    int MaxRetries = 10;
    int DelayMS = 100;

    RetryPolicy policy = new RetryPolicy<SqlAzureTransientErrorDetectionStrategy>(MaxRetries, TimeSpan.FromMilliseconds(DelayMS));
    TransactionOptions tso = new TransactionOptions();
    tso.IsolationLevel = IsolationLevel.ReadCommitted;
    policy.ExecuteAction(() =>
    {
        using (TransactionScope ts = new TransactionScope(TransactionScopeOption.Required, tso))
        {
            int cId = 29485;

            Customer c1 = (from x in dc.Customers
                           where x.CustomerID == cId
                           select x).First();

            Address ad1 = (from x in dc.Addresses
                           from y in dc.CustomerAddresses
                           where y.CustomerID == cId && x.AddressID == y.AddressID
                           select x).FirstOrDefault();

            string firstName = c1.FirstName;
            c1.FirstName = c1.LastName;
            c1.LastName = firstName;

            dc.SaveChanges();

            string addressLine1 = ad1.AddressLine1;
            ad1.AddressLine1 = ad1.AddressLine2 == null ? "dummy data" : ad1.AddressLine2;
            ad1.AddressLine2 = addressLine1;
            dc.SaveChanges();

            ts.Complete();
        }
    });
}

SQL Transaction	0 – Begin
RPC:Completed	exec sp_executesql N'SELECT TOP (1) [Extent1].[CustomerID] AS [CustomerID], …
Audit Logout
RPC:Completed	exec sp_reset_connection
RPC:Completed	exec sp_executesql N'SELECT [Limit1].[AddressID] AS [AddressID], …
Audit Logout
RPC:Completed	exec sp_reset_connection
RPC:Completed	exec sp_executesql N'update [SalesLT].[Customer] set [FirstName] = @0, [LastName] = @1 …
Audit Logout
RPC:Completed	exec sp_reset_connection
RPC:Completed	exec sp_executesql N'update [SalesLT].[Address] set [AddressLine1] = @0 …
Audit Logout
RPC:Completed	exec sp_reset_connection
SQL Transaction	1 – End
Audit Logout

Case #3 – Implement Retry Policy in OnContextCreated

If your scenario is such that the EF queries return very quickly, which is the most typical pattern, then it probably suffices to employ a retry policy solely at the connection level. If other words, only retry when a connection fails to open. The premise is that if the client application acquires a valid connection, no other network related errors will occur while during the execution of my queries. If an error does occur, the application can catch the exception and resubmit as it would in a traditional non-cloud based implementation. Besides, this tactic offers the least invasive solution because as you will from the code sample below, we only have to implement the retry policy in one spot. The remainder of the EF code works the same as if it was executing against an on premise SQL Server.

Remember earlier we stated that the ‘validity’ of a connection is only determined after a command is issued against the server. So conceivably to ensure a valid connection, one must attempt to open a connection and submit a command to the database. If an exception is thrown, the connection is close and thus removed from the pool. There is an associated overhead to opening and submitting a command just to check the validity of a connection and for this reason the ADO.NET connection pool elects not to perform this on behalf of the client application.

The tactic is to implement the OnContextCreated partial method of the models context which is called each time a new context is instantiated. In this partial method, employ a retry policy which opens a connection, submits a dummy query and handles exceptions with proper closure of invalid connections. In this way, the pool is ‘cleansed’ of all connections that have disconnected due to network glitches or idle expirations. The tradeoffs are obvious, the additional round trip to the SQL Azure database and the possible delay while the OnContextCreated method closes invalid connections.

partial void OnContextCreated()
{
    int MaxRetries = 10;
    int DelayMS = 100;

    RetryPolicy policy = new RetryPolicy<SqlAzureTransientErrorDetectionStrategy>(MaxRetries, TimeSpan.FromMilliseconds(DelayMS));
    policy.ExecuteAction(() =>
    {
        try
        {
            string ss = Connection.ConnectionString;
            Connection.Open();
            var storeConnection = (SqlConnection)((EntityConnection)Connection).StoreConnection;
            new SqlCommand("declare @i int", storeConnection).ExecuteNonQuery();
            // throw new ApplicationException("Test only");
        }
        catch (Exception e)
        {
            Connection.Close();
            throw e;
        }
    }
    );
}

RPC:Completed	exec sp_executesql N'declare @i int; set @i = @ix',N'@ix int',@ix=1
RPC:Completed	exec sp_executesql N'SELECT TOP (1) [Extent1].[CustomerID] AS [CustomerID], …
RPC:Completed	exec sp_executesql N'SELECT [Limit1].[AddressID] AS [AddressID], …
Audit Logout

The sample above submits a lightweight batch to the database, the only overhead to the database is the time to execution of the RPC statement. All looks good, but there is one important fact that may catch you. From the profiler trace we observe that the same connection is used for each query database query. This is by design and as discussed early, i.e. when a connection is explicitly opened by the developer it tells EF not to open/reopen a connection for each command. The series of Audit Login/Logout events to retrieve the customer entity or address entity are not submitted as we saw in Case #1 and #2. This means we cannot implement a retry policy for each individual query like I showed earlier. Since the EntityConnection has been assigned to the ObjectContext, EF takes the position that you really truly want to use one connection for all of your queries within the scope of that context. Retrying a query on an invalid or closed connection can never work, a System.Data.EntityCommandExecutionException will be thrown with an inner SqlException contains the message for the error.

Conclusion

Three cases were examined. Retry policies with/without a transaction scope and an implementation of OnContextCreated. The first two cases apply if you wish to introduce retry policies on all of your queries, in or out of a TransactionScope. You may find justification in this approach if your EF queries are apt to become throttled by SQL Azure which manifests itself during execution. The programming paradigms while not overly complex do require more work on the part of the developer. Applying a retry policy in the OnContextCreated method provides slimmer coverage but offers the best bang for the effort. In the majority of cases, network related exceptions tend to reveal themselves just after the connection is retrieved from the pool, thus a policy implemented right after the connection is opened should prove sufficient for those cases where database queries are expected to return quickly.

Authored By: James Podgorski
Review By: Valery Mizonov, Mark Simms, Faisal Mohamood

Distributed Cache != Database

Rama Ramani — Thu, 09 Dec 2010 23:55:05 GMT

Even though this topic looks obvious, there have been some interesting customer scenarios centered around this topic. I am hoping that this blog helps for decision making. As a basic tenet, in most solutions, a distributed cache will be used along with a database to optimize the application performance. In most cases, one would not replace one of these technologies with the other, since they provide different set of capabilities. However, based on some of the new kinds of web workload, there are some key criteria where a distributed cache can be chosen over a database.

This blog is a synopsis from a recent presentation done at SQL PASS.

Considerations

Here are a set of questions to ask yourself when looking at this decision:

Are there expensive 'key based lookup' operations?
Are there rarely changing data items accessed frequently?
Are there a lot of temporal writes?
Do you need a scalable ASP.NET session store?
Is highly availability in memory enough instead of requiring durability?

Most benefit is got when objects cached in AppFabric are frequently accessed aggregated objects, created by executing JOIN across several tables by a stored procedure or by making a set of Web service calls or a combination of both. For example, consider a popular forums website with over 300M page views per month. Each time, a user visits the forums home page, the ASP.NET application might have to run a stored procedure to aggregate the set of Posts, the various related Threads, showcase stats for the number of unanswered questions and review rating for the popular topics in each category. With 100s of categories and 1000s of Posts, each Post having 10s of Threads very quickly this becomes a scaling problem. In order to make this efficient, the aggregated "ForumPost" object can be cached so that subsequent read requests are made against a distributed cache such as AppFabric Cache thus freeing up the database server for transactional and durable data. So one will have both the distributed cache and the database, just doing different things

Storing an entire table and raw rowsets in AppFabric cache is not going to be optimal, since there are serialization and de-serialization costs when doing GETs or PUTs from AppFabric Cache. The latency of requests will not be optimal. However, depending on the how overloaded the database system gets and your allowable performance metrics, this approach might be useful. However, this is not the typical usage.

Setting up High Availability is a configuration knob in AppFabric Cache. There is no need to have any high end hardware or complex deployment techniques. And it is available at a Named Cache level and allows to apply this selectively

AppFabric Cache provides elastic scale thus allowing your data or application tier to scale linearly. Adding or Removing nodes at run-time can be done based on your needs. This is due to the scale-out architecture that it uses by leveraging some core platform components such as Fabric and CAS (Common Availability Substrate)

ASP.NET session state is one such scenario where temporal reads & writes can remain in-memory, highly available and does not really need durable storage.

Another related scenario is when performing a lot of computations (reads and writes) with the need of a "centralized scratch pad", which again may not require durability. The final result could be persisted in durable storage, like a database server

If your application needs rich querying, then the relational operators & model will work. However, if you are dealing with complex event processing with real time querying involving time windows, a product like StreamInsight may be a better fit. AppFabric Cache has support for tags and allows 'Bulk' operations which may work as a basic workaround. However this does not provider querying functionality.

Transactions and Durability are some of the core tenets in database systems. AppFabric Cache does not support them out of the box. We have got ASKs about Write-Behind feature to persist the in-memory contents and this feature is being prioritized and evaluated for a future release.

Here is a scorecard that compares a database server with AppFabric cache based on the criteria above.

Criteria	Database server	AppFabric Cache
<key, value> where value is an aggregated object
Ease of setting up HA
Ease of Scale out
ACID properties
Temporal data
Read-only data
Rich query semantics

And finally, you would have both of them in your solution architecture, possibly warming up the cache with the aggregated objects and expiring them or keeping cached objects in sync explicitly with the backend changes.

You may not agree with the * rating since it varies by scenario, but some of the aspects should be factored in your decision criteria.

Happy Caching!

Contribution from Todd Robinson is acknowledged.

Authored by: Rama Ramani
Reviewed by: Quoc Bui, Christian Martinez

Fronting long-running WF Services with MSMQ, the right way

Emil Velinov — Wed, 08 Dec 2010 14:00:00 GMT

A couple of weeks ago I was working with a customer who engaged me to assist with a "little" problem they had – under certain conditions, the system would lose messages coming from a MSMQ queue and designed to correlate to instances of a long-running WF service hosted in IIS/AppFabric.

The Rule of Thumb

Now, because I don't want to waste anybody's time, before I go into explaining all the details I'll just give you the rule of thumb from my learning: if you want to reliably expose your WF services via MSMQ, always use transactional queues coupled with a TransactedReceiveScope inside the workflow definition.

As long as you remember the above and are comfortable implementing it, you can happily stop reading right here. For the curious amongst us, I'll continue with the full story.

The Scenario

So, the customer had a long-running WF-based service that at some point in the execution was awaiting a correlated message to come in, via an endpoint bound to a non-transactional MSMQ queue.

(By the way, if you are looking for information on enabling MSMQ communication for WF services, I would point you to an excellent step-by-step guide written by Cindy Song and available here.)

Anyway, under normal circumstances, everything works OK, with the correlated message consumed off the queue and fed into the right workflow instance, which would continue its execution as expected. However, as with any real-life solution, the customer had a few scenarios (not very uncommon) that pushed the envelope a bit – welcome to the real world! So, let's look at the easiest and most obvious to explain in a blog post – a workflow instance starts, sends a one-way request to an external application or a human to perform some task, continues to do a bit of work, and then starts waiting for the external one-way "response" to come in (via a non-transactional MSMQ queue) before completing. Conceptually, this is depicted below:

Now, let's look at what happens if one of the "Do something" activities fails – obviously the workflow instance will get suspended and never reach the "Receive TaskCompleted" activity. Unless you've created some special logic on the client to detect such conditions the following sequence will happen:

The client "Complete Task" message will go to the queue bound to the "Receive TaskCompleted" endpoint
From there, the workflow endpoint using the MSMQ binding will pick up the message (this takes the message off the queue completely and unconditionally)
The message dispatcher, which is the WF runtime component responsible for inspecting incoming messages and matching them to their corresponding WF instances, will try to get to our now suspended WF instance, which of course will fail.

What now? The message has been consumed from the queue and because it is a non-transactional queue there is no rollback. At the same time the target WF instance is incapable of receiving the message and AppFabric itself does not provide a temporary message store to put the message into. So, at that point, the runtime's only choice is to logs an exception in the AppFabric monitoring DB and discontinue further actions for this message, as follows:

The bottom line is that in cases like this we are losing the message payload with no ability to handle the message manually or through a different component.

Once you realize what's happening in step 3 above, even if it is not the desired/ideal outcome, at least it is all logical and makes sense. So, let's focus on the solution - to transaction-enable the queue and the corresponding receive activity within the workflow so that if the WF runtime cannot deliver the message to an instance, the pick-up from the queue would roll back and a delivery retried later, if so configured.

Using a transactional MSMQ queue

To enable transactions for a MSMQ queue, you will need to create the queue as Transactional. What I mean here is that you cannot just change a non-transactional queue to a transactional one without re-creating it completely.

To create a transactional queue, use the Computer Management console, right-click on the "Private Queues" node, and select "New->Private Queue" from the context menu:

Make sure you name the queue after the relative URL of your WF service – also note the forward slash in the path (the service in this example is deployed to http://localhost/MsmqWF/Service1.xamlx). Then select the "Transactional" checkbox.
Once the queue is created, adjust its permissions so that the account used by the app pool running your WF service has read/write access to the queue (for simplicity, the screenshot below grants "Full Control" to all Authenticated Users):
You can also elect to enable the Journal via the General properties tab, which will provide traceability for your messages:

Configuring the service to receive messages from the transactional queue

The first step is to update the workflow to use the TransactedReceiveScope for all MSMQ-bound Receive activities. Let's assume we have the following super-simple workflow with just two one-way Receive activities - Start, and FollwowUp, which is correlated to the Start:

We will need to modify the definition to use the TransactedReceiveScope activity as depicted below:

As we can see, the change is as simple as placing the FollowUp activity into the Request placeholder of a TransactedReceiveScope activity.

The second step is to create the correct web.config file entries for the service's endpoint(s), along with their binding configuration. For the sample service below, my updated web.config file has the following content:

The second, MSMQ-bound, endpoint is using the netMsmqBinding, with a custom configuration named netMsmqBinding_Config, which specifies the exactlyOnce attribute with a value of true (meaning it should use transactional semantics for communicating with the queue). The custom binding configuration also specifies what retry semantics should be used for "failed" messages via the receiveRetryCount, maxRetryCycles, and retryCycleDelay attributes. The meaning and default values for these attributes and how they affect the MSMQ message delivery retry logic can be found on MSDN here. With the sample web.config above, if the target WF instance cannot receive the request, the message will be placed in the "retry" pool of the queue and a single re-delivery attempt will be made after 1 minute. The receiveErrorHandling attribute with a value of Move means that if all specified delivery attempts fail, the message will be moved to the poison queue where another process or a person can handle the error condition (the topic of poison queues is also covered in the same MSDN article).

The Conclusion

When using MSMQ to communicate with long-running WF services, the system design should always employ transactional queues for reliable delivery. This will effectively eliminate the potential for message loss in the WF message dispatcher in cases where the correlated target WF instance cannot either be found, activated, or capable of receiving the incoming message at the given point in time.

Thanks for reading again and happy MSMQ'ing! J

Authored by: Emil Velinov
Reviewed by: Keith Bauer, Christian Martinez

StreamInsight: Reading from other data contexts into StreamInsight with LINQPad

Mark Simms [MSFT] — Sat, 04 Dec 2010 03:46:00 GMT

The LINQPad driver for StreamInsight provides an awesomely easy way to run StreamInsight queries using easily accessible data contexts of temporal streams. However, as LINQPad only supports a single active data context connection in a query it can be difficult to use one of the hard to read “without a data context” sources (such as OData) and use StreamInsight.

As LINQPad does enable execution of generic .NET code, we can use a little trick to get around the single-connection context limitation simply placing the embedded server creation code (i.e. Server.Create()) directly into a LINQPad query statement (regardless of the active context).

Step 1: Reading an OData Service in LINQPad

We’ll start off by importing the Northwind data set as an OData service, then converting the Orders table into a CepStream. For more background on how to work with the new (as of 1.1) IEnumerable/IObservable adapters for StreamInsight refer to Colin Meek’s excellent StreamInsight Sequence Integration: Five Easy Pieces post.

Start a copy of LINQPad with the StreamInsight driver installed. Refer to the StreamInsight LINQPad Driver post if you haven’t yet done this.
From the data context list, click on Add Connection. From the Choose Data Context dialog, select WCF Data Services (OData) and click Next.

Type the URI http://services.odata.org/Northwind/Northwind.svc into the LINQPad connection dialog, and click OK.

The Northwind OData context is now available for use in LINQPad. One question that might immediately crop up is why didn’t I use the StreamInsight database context instead of the OData context? LINQPad, unfortunately, doesn’t yet support multiple database contexts in a single query. With the need to pull data from one context (LINQ to OData) and process it in another (StreamInsight), and the LINQ to OData context being rather challenging to create on the fly, we’ll use the Northwind as our primary context, and dynamically generate a StreamInsight embedded server.

The Northwind data context should now be available in the list of connections. From the query pane, select C# Statement(s) as the language, and the Northwind URI as the database. We use C# Statement instead of C# Expression to let us build up more complicated objects and code (such as the StreamInsight embedded server object).

Let’s run a simple query to make sure that the data feed is flowing through appropriately. Type the code in the block below into the query window and press F5 to execute the query.

// Retrieve orders from the Northwind OData feed that 
// have valid shipping dates and order dates.  Select 
// fields to be used in the StreamInsight query (basic 
// types only) 
var orders = from o in Orders 
                where o.ShippedDate != null && o.OrderDate != null 
                orderby o.OrderDate ascending 
                select new 
                { 
                    CustomerName = o.Customer.CompanyName, 
                    OrderDate = (DateTime)o.OrderDate, 
                    ShippedDate = (DateTime)o.ShippedDate, 
                    Region = o.ShipRegion 
                }; 
  
orders.Dump(); 

The results should be similar to those shown in the screenshot below. Once you’ve validated that the OData feed is being consumed properly, delete the orders.Dump() call.

IOrderedQueryable<> (200 items)
CustomerName	OrderDate	ShippedDate	Region
Vins et alcools Chevalier	7/4/1996 0:00	7/16/1996 0:00	null
Toms Spezialitäten	7/5/1996 0:00	7/10/1996 0:00	null
Hanari Carnes	7/8/1996 0:00	7/12/1996 0:00	RJ
Victuailles en stock	7/8/1996 0:00	7/15/1996 0:00	null
Suprêmes délices	7/9/1996 0:00	7/11/1996 0:00	null
Hanari Carnes	7/10/1996 0:00	7/16/1996 0:00	RJ
Chop-suey Chinese	7/11/1996 0:00	7/23/1996 0:00	null
Richter Supermarkt	7/12/1996 0:00	7/15/1996 0:00	null
Wellington Importadora	7/15/1996 0:00	7/17/1996 0:00	SP
HILARION-Abastos	7/16/1996 0:00	7/22/1996 0:00	Táchira

Step 2: Creating a StreamInsight host and a basic query

Now that we have the OData feed available in LINQPad, let’s go ahead and set things up to allow us to create and execute StreamInsight queries without a data context (a data context or data connection makes this a lot easier – but since we also need the Northwind OData context, one of these has to be manually defined).

In LINQPad, click on the Query menu item, then on Query Properties (or press F4).
In the Query Properties dialog, add the following references to the Additional References list, as per the screenshot below.
- Microsoft.ComplexProcessing.dll
- Microsoft.ComplexProcessing.Observable.dll

Click on the Additional Namespace Imports tab, and add the following namespaces as per the screenshot below.
- Microsoft.ComplexEventProcessing
- Microsoft.ComplexEventProcessing.Linq

Since this won’t be the only time (I hope ) that we use this technique, go ahead and click on the Set as default for new queries button, click OK to acknowledge the popup, then click OK to close the Query Properties dialog.
Now we can go ahead and define an embedded StreamInsight server to process our queries, by pasting the code below into the LINQPad query window.

using (Server server = Server.Create("Default")) 
{ 
    Application app = server.CreateApplication("test"); 
    
    
} 

Using the techniques described in Colin’s excellent post on using Sequences with StreamInsight v1.1, let’s convert the set of orders into a temporal stream. The code below demonstrates how to obtain a CepStream<> of intervals from the enumerable set of orders.

// Convert the set of orders into an interval stream 
var orderStream = orders.ToIntervalStream(app, t => 
    IntervalEvent.CreateInsert((DateTimeOffset)t.OrderDate, (DateTimeOffset)t.ShippedDate, t), 
    AdvanceTimeSettings.IncreasingStartTime); 

Now we’re ready to write our basic query. Let’s look for the condition wherein we have more than one outstanding order in a region (i.e. if the order date and ship dates overlap).

// Group the order stream by region, and create 
// snapshot windows to look at the sets of overlapping 
// orders.  Count the number of orders.    
var regionCounts = from o in orderStream 
                   group o by o.Region into regionGroups 
                   from win in regionGroups.SnapshotWindow(SnapshotWindowOutputPolicy.Clip) 
                   select new { ShipRegion = regionGroups.Key, Count = win.Count() }; 
  
// Look for regions which have more than 1 overlapping 
// order 
var filter = from e in regionCounts 
             where e.Count > 2 
             select e; 
  
// Dump the results of the query out into an enumerable 
var snk = from i in filter.ToIntervalEnumerable() 
          where i.EventKind == EventKind.Insert 
          select new { i.StartTime, i.EndTime, i.Payload.ShipRegion, i.Payload.Count }; 
snk.Dump(); 

Let’s run the query (by pressing F5) and observe the results.

IEnumerable<> (130 items)
StartTime	EndTime	ShipRegion	Count
7/8/1996 7:00:00 AM +00:00	7/9/1996 7:00:00 AM +00:00	null	3
7/9/1996 7:00:00 AM +00:00	7/10/1996 7:00:00 AM +00:00	null	4
7/10/1996 7:00:00 AM +00:00	7/11/1996 7:00:00 AM +00:00	null	3
7/11/1996 7:00:00 AM +00:00	7/12/1996 7:00:00 AM +00:00	null	3
7/12/1996 7:00:00 AM +00:00	7/15/1996 7:00:00 AM +00:00	null	4

Interesting result combining LINQPad exploration of a LINQ to WCF OData and StreamInsight – but what went into producing that result? Which collection of events resulted in that output event? Answering that question will be my next post on using the Event Flow Debugger with LINQPad and StreamInsight

// Retrieve orders from the Northwind OData feed that 
// have valid shipping dates and order dates.  Select 
// fields to be used in the StreamInsight query (basic 
// types only) 
var orders = from o in Orders 
where o.ShippedDate != null && o.OrderDate != null 
orderby o.OrderDate ascending 
select new 
{ 
    CustomerName = o.Customer.CompanyName, 
    OrderDate = (DateTime)o.OrderDate, 
    ShippedDate = (DateTime)o.ShippedDate, 
    Region = o.ShipRegion 
}; 
  
using (Server server = Server.Create("Default")) 
{ 
    Application app = server.CreateApplication("test"); 
  
    // Convert the set of orders into an interval stream 
    var orderStream = orders.ToIntervalStream(app, t => 
      IntervalEvent.CreateInsert((DateTimeOffset)t.OrderDate, (DateTimeOffset)t.ShippedDate, t), 
      AdvanceTimeSettings.IncreasingStartTime); 
    
    // Group the order stream by region, and create 
    // snapshot windows to look at the sets of overlapping 
    // orders.  Count the number of orders.    
    var regionCounts = from o in orderStream 
        group o by o.Region into regionGroups 
        from win in regionGroups.SnapshotWindow(SnapshotWindowOutputPolicy.Clip) 
        select new { ShipRegion = regionGroups.Key, Count = win.Count() }; 
  
    // Look for regions which have more than 1 overlapping 
    // order 
    var filter = from e in regionCounts 
        where e.Count > 2 
        select e; 
  
    // Dump the results of the query out into an enumerable 
    var snk = from i in filter.ToIntervalEnumerable() 
          where i.EventKind == EventKind.Insert 
          select new { i.StartTime, i.EndTime, i.Payload.ShipRegion, i.Payload.Count }; 
    snk.Dump(); 
} 

StreamInsight: Where did that query result come from (using the Event Flow Debugger in LINQPad)?

Mark Simms [MSFT] — Thu, 02 Dec 2010 17:24:00 GMT

Occasionally, when working with StreamInsight queries an unusual result will crop up, leading to asking the rather blunt question “where did that come from?”. Luckily the StreamInsight Event Flow Debugger has a great feature (Root Cause Analysis) that will let us examine how certain results were generated.

If you’ve never used the Event Flow Debugger, or are having trouble working with it, please have a look through StreamInsight: Getting started with using the (Event Flow) debugger, viewing diagnostics, and exposing the management service.

We’ll start by creating a system that produces “questionable” results, then examining it with the Event Flow Debugger. This system will:

Pull in a stream of events from the Northwind OData feed
Look for a specific condition occuring in the Orders table (thanks to Allan Mitchell for the query idea ), using LINQPad and the StreamInsight driver to execute the query

If you haven’t installed LINQPad with the StreamInsight driver, the step-by-step instructions may be found at StreamInsight LINQPad Driver on the StreamInsight team blog. This tool combination makes ad-hoc querying drastically simpler, and I highly recommend dropping down the $30 for autocompletion. Well worth it!

With the system up and running, we’ll use the offline trace file capabilities of StreamInsight to capture events flowing through the system, then analyze the results to see if it “makes sense”. The details of using LINQPad to pull an OData stream from the internet and query it with StreamInsight are covered StreamInsight: Reading from other data contexts into StreamInsight with LINQPad. The trace file that I use in this blog post was generated by running the LINQ statement from that blog post.

Capturing a trace

We’ll use the offline trace capability of StreamInsight to figure out some root cause analysis of output events. From the previous post, let’s track down what went into producing this output event:

7/12/1996 7:00:00 AM +00:00

7/15/1996 7:00:00 AM +00:00

null

Open up a command prompt, and navigate to the C:\Program Files\Microsoft StreamInsight 1.1\Tools directory.
From this directory, use trace.cmd to start capturing tracing information via trace start rootcause.etl.
From LINQPad, execute the query again.
From the command prompt stop tracing by executing the command trace stop.
If you observe the contents of the C:\Program Files\Microsoft StreamInsight 1.1\Tools directory the rootcause.etl file should be visible with a non-zero length. We’ll now use the Event Flow Debugger to perform a root cause analysis.

Using the Event Flow Debugger against a Trace file

Start the Event Flow Debugger (Start –> Microsoft StreamInsight 1.1 x64 (en) –> Event Flow Debugger).
Click on File, then on Open.
Browse to the C:\Program Files\Microsoft StreamInsight 1.1\Tools directory. Select the rootcause.etl file and click Open.
From the Object Explorer, expand the rootcause.etl node, then the Server node, then the test application node.
Right-click on the IEnumerable subscription 0 query, and click Display Query.

If we expand the IEnumerable stream 1 input adapter, we’ll see all of the events that flowed into this query. We’ll come back to this in a bit, but we’re more interested right now in looking at the output and determining root cause.

Scroll down to the OutputAdapter_Cleanse box, and expand it. Since we’re (for this exploration) only interested in the actual output events (rather than the liveliness of the query by examining CTI’s), let’s go ahead and filter for INSERT events by adding a filter to the operator view. This filter will be EventKind == “Insert”.

We’re trying to track down what went into producing a specific event, so let’s add a more detailed filter to the cleanse box. In this case we added “&& StartTime == DateTime.Parse(“7/12/1996 7:00:00 AM”)”

Now that we’ve tracked down the specific event we’re interested in, let’s go ahead and see what went into producing it. Click on the event to select it, then click Analysis in the menu, and finally click Root Cause Analysis.

Scroll back up to the input adapter, and we can now observe the four events that went into generated our output event!

Takeaways

LINQPad, together with the StreamInsight LINQpad driver makes a great ad-hoc query and exploratory interface.
Use the offline trace viewing capability of the Event Flow Debugger to capture query activity and investigate how your output events came to be.

Basics on how to setup an App Fabric Cache server and API usage

Jaime Alva — Wed, 01 Dec 2010 02:02:00 GMT

At this year’s SQL PASS summit, I had the opportunity to talk with customers about App Fabric Cache and in preparation for the event, I put a demo to show some of the characteristics of the available APIs. The following is an explanation on how to setup the demo, what to expect when running it (as if you had attended), along with notes based on the most frequently asked questions.

The setup

For simplicity, everything will be contained on one machine. Start by downloading the appropriate Windows Server AppFabric bits and installing them and make sure to add the Caching Services, Cache Client and Cache Administrator (I would recommend installing everything so as to play with the host bits as well but nothing of that is included in this blog), see below.

After the bits are installed the configure Hosting Services window will come up, again, for simplicity, take the defaults as shown

Set the cache host with a XML configuration provider (SQL provider not use only for simplicity, SQL is however the recommended setup) and a small cluster. Make sure the service account has rights to the XML share, as follows. If you have ran the wizard on the same share before you will have to remove the existing ClusterConfig.xml file (or rename it)

Leave all ports as per defaults (see below).

for more installation details see Install Windows Server AppFabric and Configure Windows Server AppFabric.

NOTE: Do not try to setup either a client or server by copying and GACing the App Fabric Cache DLLs, this will lead to errors and is not a supported installation.

Now that the cache host is setup and configured, run the “Caching Administration Windows PowerShell” in elevated rights, as an Administrator. And to start the cache host, run the following command.

>Start-CacheHost

You will be prompted for the hostname and cache port. Give the machine name and port 22233 (since the defaults were taken when running the configuration wizard), you should see the following

HostName : CachePort Service Name Service Status Version Info

-------------------- ------------ -------------- ------------

jaimelap2:22233 AppFabricCachingService UP 1 [1,1][1,1]

Setup your user account to an allowed client user, in my case I had

>Grant-CacheAllowedClientAccount redmond\jaimeab

Create the cache for the demo application and look at the current statistics

>new-cache default

>get-cachestatistics default

You will see the following stats since nothing has yet taken place:

Size : 0

ItemCount : 0

RegionCount : 0

RequestCount : 0

MissCount : 0

Running the demos

Here is the demo project (built in VS2010 ultimate), make sure the two following references are correctly set, otherwise your setup may be incorrect, refer to this blog for troubleshooting hints.

Running the solution will show the following windows form:

Demo #1, Basic Object Addition, Retrieval and Removal

This is the most commonly use APIs, as per the message in the Status text start by adding an object,

first start by clicking the Add button, this will add a single entry into the cache, the status will show “>>Add-Object Added to Cache [key=LockingDemo]”.
Clicking the Get button will show the content of the added object which was retrieved by leveraging the Key “lockingDemo”, the status will show “>> Get Object and reference retrieved from cache. [key=LockingDemo, Object=Object stored in cache]”

At this point go back to the “Caching Administration Windows PowerShell” console and type:

>Get-CacheStatistics default

The RequestCount will be 1, this is the new added object had just got added

Hit the Remove button and then the Get button, you will see the fail status as it indicates that no object was found. The same can be seen if an object is added and retrieved after 15 minutes, which is the default time to keep an item in the cache, this is further explore in the Demo #4 below.

The Add API can be thought as a type of insert statement but it only takes a key value pair. Note that the Get API merely retrieved whatever it found in the object in an optimistic manner, it did not care if it was changed by any other process, as it retrieved the latest version.

Demo #2, Version retrieval, GetIfNewer and Put

If you want to know when an object has changed from the last time it was retrieved then GetIfNewer is a good option. Looking at the code in the solution, when the Get API was executed (in line 134 of form1.cs) a DataCacheItemVersion was also collected, this can be thought of as a type of version for the gotten item. See line 143 in Form1.cs

RetrievedStrFromCache = (string)myDefaultCache.Get(myKey, out myVersionBeforeChange)

For simplicity, say it is retrieved as Version1, if another process comes and does a Put against the same item (by using the same key) then it will change its version to say Version2, in the same way the DataCacheItemVersion will get updated. GetIfNewer can be used to leverage this type of versioning, by using the DataCacheItemVersion retrieved in the Get API in the GetIfNewer API, it will know to return a NULL object if it finds an object with the same DataCacheItemVersion in cache, otherwise it will return the new (updated) object.

Go back to the demo form,

click on the Add button (if a valid object is still present the Object already exist exception will be raised)
Then click on the GetIfNewer button and you will get a message saying that a NULL object was retrieved.
Now click on the Put button (which can be thought as an update statement)
Go back and click on the GetIfNewer button, now you will get the object that the Put API just updated, the value string changed from “Object stored in cache” to “MOD 1 - Object stored in cache”. Note that clicking on the GetIfNewer button again will return a NULL item since the version has not changed from the last time the API was executed. This also shows that the value of the DataCacheItemVersion also gets updated with each execution of the GetIfNewer API.

Similarly, if GetIfNewer is executed after doing a Remove and an Add, it will also retrieve an object since these two operations (Remove and Add) will have, in effect, modified the object.

Demo #3, Contention, GetAndLock, Unlock and PutAndUnlock

As explained, the Get API does an optimistic retrieval but there are cases when a pessimistic retrieval may be preferred. This scenario will be most adequately use for Resource data (shared reads & writes across multiple users). For those cases, the GetAndLock API will allow locking an object for a given period of time; the demo does it for 5 seconds. Two other things to keep in mind are the key and the DataCacheLockHandle, as these are needed to handle the locking semantics.

Running the demo:

Click the Add button, if an object already exist an exception will be shown.
Click on the GetAndLock button, this will collect a DataCacheLockHandle object that will be required to gracefully unlock the object (the Put API may also do this without requiring a DataCacheLockHandle, more on this further below).
Click the GetAndLock button again, an exception explaining that the object is already lock will be generated.
Before the 5 seconds for expiration have taken place, hit the Unlock (Bad) button, this will generate an exception due to the fact that even though the correct key is used, the incorrect DataCacheLockHandle is employed.
If the GetAndLock button is press right after (before the 5 seconds lock is reached) then the exception showing that the object is still lock will be shown, this since the previous unlock called was unsuccessful.
When the Unlock (ok) button is pressed, the object will be gracefully unlock, as this method uses the correct key and DataCacheLockHandle
Pressing the GetAndLock button will not show an exception but instead will lock the object again for 5 seconds.

The PutAndUnlock button will also correctly unlock the object but, as stated by its name, it will update the object at the same time.

NOTE: the Put button will override the lock semantics; that is, the object will be unlock and update without requiring a DataCacheLockHandle. This is because the intent with the lock APIs is for all applications to honor it, so in order for the locking to be honored, it is incumbent upon the clients to be using the Lock methods since not using them will override the locking guarantees.

Demo #4, Enable Local cache

This demo goes over the ability to locally cache objects, this will save a round trip to the cache cluster, and in essence, it will eliminate the network delay to and back from the cache server.

To start the demo start the “Caching Administration Windows PowerShell” tool and run the following commands

>restart-cachecluster

>get-cachestatistics default

The purpose here is to clean all the objects currently cache; the statistics results should show all zeros.

Running the demo:

Click the Add button, as seen in Demo #1, after adding an object the statistics show a RequestCount of 1.
Choose Enabled from the local cache drop down
Click on the Get button a few times.
Go back to the “Caching Administration Windows PowerShell” tool and run the “Get-cacheStatistics default” command again. The requestCount should still stay at 1, this is because the configuration provider did not register the request since it was locally served.

Note that if you do a GetIfNewer instead of a Get, the requestCount will increase; this is because GetIfNewer cannot be locally supplied, as it needs to check with all the other cache hosts (in this case, the XML configuration provider).

To achieved the demonstrated, two DataCacheFactories are leverage - see line 54~59 in form1.cs. The DataCacheFactory named “myCacheFactory” is the one use for non-local cache and the one named “myCacheFactory_LC” is the one setup for local cache, its configuration property is defined in line 57, as follows.

//Create the local cache cacheFactory constructor, kept for 5 min.

configuration.LocalCacheProperties = new DataCacheLocalCacheProperties(1000, new TimeSpan(0, 5, 0), DataCacheLocalCacheInvalidationPolicy.TimeoutBased);

Note that the timeout is setup to 5 minutes, to demo it,

Go back to the “Caching Administration Windows PowerShell” tool and run the “Get-cacheStatistics default” and make note of the requestCount
Makes sure the local cache drop down is set to Enabled
Click the Add button
Wait for 5 minutes and then hit the Get button
Go back to the “Caching Administration Windows PowerShell” tool and run the “Get-cacheStatistics default” again. The requestCount will have gone up, as the local cache item would have expired and had to be retrieved from the host cache.

Reviewed by: Christian Martinez and Mark Simms

Hybrid Cloud Solutions With Windows Azure AppFabric Middleware

SurenM — Tue, 30 Nov 2010 06:28:00 GMT

Abstract

(updated version)

Technical and commercial forces are causing Enterprise Architects to evaluate moving established on-premises applications into the cloud – the Microsoft Windows Azure Platform.

This blog post will demonstrate that there are established application architectural patterns that can get the best of both worlds: applications that continue to live on-premises while interacting with other applications that live in the cloud – the hybrid approach. In many cases, such hybrid architectures are not just a transition point — but a requirement since certain applications or data is required to remain on-premises largely for security, legal, technical and procedural reasons.

The cloud is new, and the hybrid cloud is even newer. There are many technologies that have just been released or announced so there is no one source for authentic information, especially one that compares, contrasts, and ties it all together. This blog, and a few more that will follow, is an attempt to demystify and make sense of it all. We begin with a brief review of two prevalent deployment paradigms and their influence on architectural patterns: On-premises and Cloud. After that, we discuss developing the hybrid architecture.

This posting takes an architect’s perspective and surveys the major building block components that compose this hybrid architecture. We also match requirements against the capabilities of available and announced Windows Azure and Windows Azure AppFabric technologies. Our discussions also factor in the usage costs and strategies for keeping these costs in check. We conclude with a survey of interesting and relevant Windows Azure technologies announced at Microsoft PDC 2010 - Professional Developer's Conference during October 2010.

On-Premises Solutions

On-premises solutions are usually designed to access data repositories (through client or browser applications) and services hosted inside of a corporate network. In the graphic below, Web Server and Database Server are within a corporate firewall. The user, via the browser, is authenticated and has access to data.

Figure 1: Typical On-Premises Solution (Source - Microsoft.com)

The Patterns and Practices Application Architecture Guide 2.0 provides an exhaustive survey of the on-premises Applications. This Architecture Guide is a great reference document and helps you understand the underlying architecture and design principles and patterns for developing successful solutions on the Microsoft application platform and the .NET Framework. The Microsoft Application Platform is composed of products, infrastructure components, run-time services, and the .NET Framework, as detailed in the following table (source: .NET Application Architecture 434 Guide, 2nd Edition).

Table 1: Microsoft Application Platform (Source - Microsoft.com)

The on-premises architectural patterns are well documented (above), so in this blog post we will not dive into prescriptive technology selection guidance here, however let’s briefly review some core concepts since we will reference them while elaborating hybrid cloud architectural patterns.

Hosting The Application

Commonly* on-premises applications (with the business logic) run on IIS (as an IIS w3wp process) or as a Windows Service Application. The recent release of Windows Server AppFabric makes it easier to build, scale, and manage Web and Composite (WCF, SOAP, REST and Workflow Services) Applications that run on IIS.

* As indicated in Table 1 (above) Microsoft BizTalk Server is capable of hosting on-premises applications; however, for sake of this blog, we will focus on Windows Server AppFabric

Accessing On-premises Data

Your on-premises applications may access data from the local file system or network shares. They may also utilize databases hosted in Microsoft SQL Server or other relational and non-relational data sources. In addition, your applications hosted in IIS may well be leveraging Windows Server AppFabric Cache for your session state, as well as other forms of reference or resource data.

Securing Access

Authenticated access to these data stores are traditionally performed by inspecting certificates, user name and password values, or NTLM/Kerberos credentials. These credentials are either defined in the data source themselves, heterogeneous repositories such as in SQL Logins, local machine accounts, or in directory Services (LDAP) such as Microsoft Windows Server Active Directory, and are generally verifiable within the same network, but typically not outside of it unless you are using Active Directory Federation Services - ADFS.

Management

Every system exposes a different set of APIs and a different administration console to change its configuration – which obviously adds to the complexity; e.g., Windows Server for configuring network shares, SQL Server Management Studio for managing your SQL Server Databases; and IIS Manager for the Windows Server AppFabric based Services.

Cloud Solutions

Cloud solutions typically access local resources and services, but they can eventually interact with remote services running on-premises or in the cloud. Cloud applications usually hosted by a ‘provider managed’ runtime environment that provides hosting services, computational services, storage services, queues, management services, and load-balancers. In summary, cloud applications consist of two main components: those that execute application code and those that provide data used by the application. To quickly acquaint you with the cloud components, here is a contrast with popular on-premises technologies:

Table 2: Contrast key On-Premises and Cloud Technologies

Category	On-Premises	Cloud
Application Hosting	Windows Server	Windows Azure
Message Queue	MSMQ	Azure Queue
Unstructured Data Store	Windows Folder Windows Drives	Azure Blobs Azure Table Azure Drive (XDrive)
Relational Data Store	SQL Server	SQL Azure
Application Cache	AppFabric Cache	Azure AppFabric Cache

NOTE: This table is not intended to compare and contrast all Azure technologies. Some of them may not have an equivalent on-premise counterpart.

The graphic below presents an architectural solution for ‘Content Delivery’. While Content creation and its management is via on-premises applications; the content storage (Azure Blob Storage) and delivery is via Cloud – Azure Platform infrastructure. In the following sections we will review the components that enable this architecture.

Figure 2: Typical Cloud Solution

Running Applications In Windows Azure

Windows Azure is the underlying operating system for running your cloud services on the Windows Azure AppFabric Platform. The three core services of Windows Azure in brief are as follows:

Compute: The compute or hosting service offers scalable hosting of services on 64-bit Windows Server platform with Hyper-V support. The platform is virtualized and designed to scale dynamically based on demand. The Azure platform runs web roles on Internet Information Server (IIS) and worker roles as Windows Services.
Storage: There are three types of storage supported in Windows Azure: Table Services, Blob Services, and Queue Services. Table Services provide storage capabilities for structured data, whereas Blob Services are designed to store large unstructured files such as videos, images, batch files in the cloud. Table Services are not to be confused with SQL Azure; typically you can store the high-volume data in low-cost Azure Storage and use (relatively) expensive SQL Azure to store indexes to this data. Finally, Queue Services are the asynchronous communication channels for connecting between Services and applications not only in Windows Azure but also from on-premise applications. Caching Services, currently available via Windows Azure AppFabric LABS, is another strong storage option.
Management: The management service supports automated infrastructure and service management capabilities to Windows Azure cloud services. These capabilities include automatic and transparent provisioning of virtual machines and deploying services in them, as well as configuring switches, access routers, and load balancers.

A detailed review on the above three core services of Windows Azure is available in the subsequent sections. Application code execution on Windows Azure is facilitated by Web and Worker roles. Typically, you would host Websites (ASP.NET, MVC2, CGI, etc.) in a Web Role and host background or computational processes in a Worker role. The on-premises equivalent for a Worker role is a Windows Service. This architecture leaves a gray area — where should you host WCF Services? The answer is – it depends! Let me elaborate. When building on-premises services, you can host a WCF Service in a Windows Service. For example, in BizTalk, a WCF Receive Location can be hosted within an in-process host instance that runs as a Windows Service). In Windows Azure you can host a WCF Service as Worker role. While you can host a WCF Service (REST or SOAP) in either a Worker Role or Web role, you typically host these services in a Web role; the one exception is when your WCF service specifically needs to communicate via TCP/IP endpoints. Web roles are capable of exposing HTTP and HTTP(S) endpoints; Worker roles add the ability to expose TCP endpoints.

Storing & Retrieving Cloud Hosted Data

Applications that require data storage in the cloud can leverage a variety of technologies including SQL Azure, Windows Azure Storage, Azure AppFabric Cache, Virtual Machine (VM) instance local storage, Instance Local Storage and Azure Drive (XDrive). Determining the right technology is a balancing exercise in managing trade-offs, costs, and capabilities. The following sections will attempt to provide prescriptive guidance on what usage scenario each storage option is best suited for. The key takeaway is this: whenever possible, co-locate data and the consuming application. Your data in the cloud, stored in SQL Azure or Windows Azure Storage, should be accessible via the ‘local’ Azure “fabric” to your applications. This has positive performance and a beneficial cost model when both the application and the data live within the same datacenter. Co-location is enabled via the 'Region' you choose for the Microsoft Data Center.

SQL Azure

SQL Azure provides an RDBMS in the cloud, and functions for all intents and purposes similarly to your on-premises version of SQL Server. This means that your applications can access data in SQL Azure using the Tabular Data Stream (TDS) protocol; or, in other words, your application uses the same data access technologies (e.g. ADO.NET, LINQ to SQL, EF, etc.) used by on-premises applications to access information on a local SQL Server. The only thing you need to do is change the SQL Client connection strings you’ve been using. You just have to change the values to point to the server database hosted by SQL Azure. Of course I am glossing over some details like the SQL Azure connection string which contains other specific settings such as encryption, explicit username (in very strict format username@servername) and password – I am sure you get the drift here. SQL Azure Labs demonstrate an alternative mechanism for allowing your application to interact with SQL Azure via OData. OData provides the ability to perform CRUD operations using REST and can retrieve query results formatted as AtomPub or JSON. Via .NET programming, you can interact at a higher level using a DataServicesContext and LINQ.

Securing Access

When using TDS, access is secured using SQL Accounts (i.e., your SQL Login and password must be in the connection string), however there is one additional layer of security you will need to be aware of. In most on-premises architectures, your SQL Server database lives behind a firewall, or at most in a DMZ, which you would rarely expose. However, no matter where they are located, these repositories are not directly accessible to applications living outside the corporate boundaries. Of course, even for on-premises solutions you can use IPSEC to restrict the range of machines that can access the SQL Server machine. Eventually, access to these databases is mediated and controlled by web services (SOAP, WCF).

What happens when your database lives on Azure, in an environment effectively available to the entire big bad world? SQL Azure also provides a firewall that you can configure to restrict access to a set of IP address ranges, which you would configure to include the addresses used by your on-premises applications. In the graphic below notice the ‘Firewall Settings’ tab where the IP Address Range is specified.

Figure 3: SQL Azure Firewall Settings

When using OData, access is secured by configuration. You can enable anonymous access whereby the anonymous account maps to a SQL Login created on your database, or you can map Access Control Service (ACS) identities to specific SQL Logins, whereby ACS takes care of the authentication. In the OData approach, the Firewall access rules are effectively circumvented because only the OData service itself needs to have access to the database hosted by SQL Azure from within the datacenter environment.

Management

You manage your SQL Azure database the same way you might administer an on-premises SQL Server database, by using SQL Server Management Studio (SSMS). Alternately, and provided you allowed Microsoft Services in the SQL Azure firewall, you can use the Database Manager for SQL Azure (codename “Project Houston”) which provides the environment for performing many of the same tasks you would perform via SSMS. The Database Manager for SQL Azure is a part of the Windows Azure platform developer portal, and provides a lightweight, web-based database management and querying service for SQL Azure databases. This allows customers to have a streamlined experience within the web browser without having to download any tools.

Figure 4: Database Manager for SQL Azure (Source - PDC 10)

One of the primary technical differences between the traditional SSMS management option is the required use of port 1433; in contrast, “Houston” is web-based and leverages port 80. An awesome demo on the capabilities of the Database Manager is available here.

Typical Usage

SQL Azure is best suited for storing data that is relational, specifically where your applications expect the database server to perform the join computation and return only the processed results. SQL Azure is a good choice for scenarios with high transaction throughput (view case studies here) since it has a flat-rate pricing structure based on the size of data stored. Additionally, query evaluation is distributed across multiple nodes.

Windows Azure Storage (Blobs, Tables, Queues)

Windows Azure Storage provides storage services for data in the form of metadata enriched blobs, dictionary-like tables, and simple persistent queues. You access data via HTTP or HTTP(S) by leveraging the StorageClient library which provides 3 different classes, respectively CloudBlobClient, CloudTableClient, CloudQueueClient), to access storage services. For example, the CloudTableClient class provides a DataServices context that enables use of LINQ for querying table data. All data stored by Windows Azure Storage can also be accessed via REST. The AppFabric CAT (team) plans to provide code samples via this blog site to demonstrate this functionality – stay tuned.

Securing Access

Access to any of the Windows Azure Storage blobs, tables, or queues is provided through symmetric key authentication whereby an account name and account key pair must be included with each request. In addition, access to the Azure blobs can be secured via the mechanism known as the Shared Access Signature.

Management

Currently there is no single tool for managing Windows Azure Storage. There are quite a few samples as well as third-party tools. The graphic below is from the Azure Storage Explorer tool, available for download via Codeplex.

Figure 5: Azure Storage Explorer (Source – Codeplex)

In general, a Windows Azure storage account, which wraps access to all three forms of storage and services, is created through the Windows Azure Portal, but is managed using the API’s.

Typical Usage

Windows Azure storage provides unique value propositions that make it fit within your architecture for diverse reasons.

Blob Storage

Charges for blob storage are based on both amount of storage used, and by the number of storage transactions, But the pricing model is designed to scale transparently to any size. This makes it best suited to the task of storing files and larger objects (high resolution videos, radiology scans, etc.) that can be cached or are otherwise not frequently accessed.

Table Storage

Table storage is billed the same way as blob storage, but its unique approach to indexing at large scale makes it useful in situations where: (a) you need to efficiently access data from datasets larger than the 50 GB per database maximum of SQL Azure; and (b) where you don’t expect to be performing joins, or are comfortable enabling the client to download a larger chunk of data and perform the joins and computations on the client side.

Queue Storage

Queues are best suited for storing pointers to work that needs to be done, due to their limited storage capacity of a maximum of 8K per message, in a manner ensuring ordered access. Often, you will use queues to drive the work performed by your Worker roles, but they can also be used to drive the work performed by your on-premises services. Bottom-line: Queues can be effectively used by a Web Role to exchange control messages in an asynchronous manner with a Worker Role running within the same application, or to exchange messages with on-premises applications

Azure AppFabric Cache

The Azure AppFabric Cache, currently in CTP/Beta as of November 2010, should give you the same high-performance, in-memory, distributed cache available with Windows Server AppFabric, as a hosted service. (The technology is in pre-release mode and hence the caveat!) Since this is an Azure AppFabric Service, you are not responsible for managing servers participating in the data cache cluster. Your Windows Azure applications can access the cached data through the client libraries in the Windows Azure AppFabric SDK.

Securing Access

Access to a cache is secured by a combination of generated authentication tokens with authorization rules (e.g., Read/Write versus Read only) as defined in the Access Control Service.

Figure 6: Access Control for Cache

Management

At this time, creation of a cache, as well securing access to it, is performed via the Windows Azure AppFabric labs portal. Subsequent to its release this will be available in the commercial Azure AppFabric portal.

Typical Usage

Clearly the cache is best suited for persisting data closer to your application, whether its data stored in SQL Azure, Windows Azure Storage, or as the result of a call to another service—or a combination of all of them. This approach is called the cache-aside model, whereby requests for data made by your application first check the cache and only query the actual data source, and subsequently add it to the cache, if it’s not present. Typically we are seeing caches used in the following scenarios:

As a scalable session store for your web applications running on ASP.NET.
As an output cache for your web application.
For objects created from resultsets from SQL or from costly web service calls, stored in cache and called from your web or worker role.
As a scratch pad in the cloud for your applications to use.

Using VM local storage & Azure Drives

When you create an instance in Windows Azure, you are creating an instance of a Virtual Machine (VM). VM, just like its on-premises counterpart, can make use of locally attached drives for storage or network shares. Windows Azure Drives are similar, but not exactly the same as, network shares. They are an NTFS formatted drive stored as a page blob in Windows Azure Storage, and accessed from your VM instance by drive letter. These drives can be mounted exclusively to a single VM instance with read/write privileges, or to multiple VM instances as read-only drives. When such a drive is mounted, it caches data read in local VM storage, which enhances the read performance of subsequent accesses for the same data.

Securing Access

An Azure Cloud Drive is really just a façade on top of a page blob, so access to the drive effectively amounts to having access to the blob store using Windows Storage Account credentials (i.e., account name and account key).

Management

For blobs, there is limited tooling for managing Azure drives outside of the StorageClient API’s. In fact, there is no portal for creating Azure Cloud drives. However, there are samples on Codeplex that help you create and manage Azure drives. The Windows Azure MMC Snap-In is available and can be downloaded from here – and the graphic below provides you a quick peek into the familiar look/feel of the MMC.

Figure 7: Windows Azure MMC Snap-In (Source - MSDN)

Typical Usage

Cloud drives have a couple of unique attributes that make them interesting choices in hybrid architecture. To begin with, if you need drive-letter based access to your data from your Windows Azure application and you want it to survive VM instance failure, a cloud drive is your best option. Beyond that, you can mount a cloud drive to instances as read-only. This enables sharing of large reference files across multiple Windows Azure roles. What’s more, you can create a VHD, for example, using Window 7’s Disk Management MMC and load your VHD to the blob storage for use by your VM instances, effectively cloning an on-premises drive and extending its use into the cloud.

Hybrid Cloud Solutions: Exposing On-premises Services to Azure Hosted Services

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. For the purposes of this article, and as illustrated in the diagram below, we will focus on the scenario where your on-premises applications are services hosted in Windows Server AppFabric and then communicate to other portions of your hybrid solution running in the cloud.

Figure 8: Typical Hybrid Cloud Solution

Traditionally, to connect your on-premises applications to the off-premises applications (cloud or otherwise), you would begin by “poking” a hole in your firewall and configuring NAT routing so that Internet clients can talk to your services directly. This approach has numerous issues and limitations, not the least of which is the management overhead, security concerns, and configuration challenges.

Connectivity

So the big question here is: How do I get my on-premises services to talk to my Azure hosted services? There are two approaches you can take: You can use the Azure AppFabric Service Bus, or you can use Windows Azure Connect.

Using Azure AppFabric Service Bus

If your on-premises solution includes WCF Services, WCF Workflow Services, SOAP Services, or REST services that communicate via HTTP(S) or TCP you can use the Service Bus to create an externally accessible endpoint in the cloud through which your services can be reached. Clients of your solution, whether they are other Windows Azure hosted services, or Internet clients, simply communicate with that end-point, and the AppFabric Service Bus takes care of relaying traffic securely to your service and returning replies to the client. The graphic below (from http://www.microsoft.com/en-us/appfabric/azure/middleware-services.aspx#ServiceBus) demonstrates the Service Bus functionality.

Figure 9: Windows Azure AppFabric Service Bus (Source - Microsoft.com)

The key value proposition of leveraging the Service Bus is that it is designed to transparently communicate across firewalls, NAT gateways, or other challenging network boundaries that exist between the client and the on-premises service. You get the following additional benefits:
The actual endpoint address of your services is never made available to clients.
You can move your services around because the clients are only bound to the Service Bus endpoint address, which is a virtual, not a physical, address.
If both the client and the service happen to be on the same LAN and could therefore communicate directly, the Service Bus can set them up with a direct link that removes the hop out to the cloud and back and thereby improves throughput and latency.

Securing Access to Service Bus Endpoints

Access to the Service Bus is controlled via the Access Control Service. Applications that use the Windows Azure AppFabric Service Bus are required to perform security tasks for configuration/registration, or for invoking service functionality. Security tasks include both authentication and authorization using tokens from the Windows Azure AppFabric Access Control service. When permission to interact with the service has been granted by the AppFabric Service Bus, the service has its own security considerations that are associated with the authentication, authorization, encryption, and signatures required by the message exchange itself. This second set of security issues has nothing to do with the functionality of the AppFabric Service Bus; it is purely a consideration of the service and its clients. There are four kinds of authentication currently available to secure access to the Service Bus:

SharedSecret, a slightly more complex but easy-to-use form of username/password authentication.
SAML, which can be used to interact with SAML 2.0 authentication systems.
SimpleWebToken, uses the OAuth Web Resource Authorization Protocol (WRAP) and Simple Web Tokens (SWT).
Unauthenticated, that enables interaction with the service endpoint without any authentication behavior.

Selection of the authentication mode is generally dictated by application connecting to the Service Bus. You can read more on this topic here.

Cost Considerations

Service Bus usage is charged by the number of concurrent connections to the Service Bus endpoint. This means that when an on-premises or cloud-hosted service registers with the Service Bus and opens its listening endpoint, it is considered as one connection. When a client subsequently connects to that Service Bus endpoint, it’s counted as a second connection. One very important point falls out of this that may affect your architecture: you will want to minimize concurrent connections to the Service Bus in order to keep the subscription costs down. It may be more likely that you would want to use the Service Bus in the middle tier more like a VPN to on-premises services rather than allowing unlimited clients to connect through the Service Bus to your on-premises service. To reiterate, the key value proposition of the Service Bus is Service Orientation; it makes it possible to expose application Services using interoperable protocols with value-added virtualization, discoverability, and security.

Using Windows Azure Connect

Recently announced at PDC 10, and expected for release by the end of 2010, is an alternative means for connecting your cloud services to your on-premises services. Windows Azure Connect effectively offers IP-level, secure, VPN-like connections from your Windows Azure hosted roles to your on-premises services. This service is not available yet and pricing details have not been released. From the information available to date, you can conclude that Windows Azure Connect would be used for connecting middle-tier services, rather than public clients, to your on-premises solutions. While the Service Bus is focused on connectivity for on-premise Services exposed as Azure endpoints without having to deal with firewall and NAT setup, Windows Azure Connect provides broad connectivity between your Web/Worker roles and on-premises systems like SQL Server, Active Directory or LOB applications.

Securing the solution

Given the distributed nature of the hybrid (on-premises and cloud) solution, your approach to security should match it - this means your architecture should leverage Federated Identity. This essentially means that you are outsourcing authentication and possibly authorization. If you want to flow your authenticated on-premises identities, such as domain credentials, into Azure hosted web sites or services, you will need a local identity (akin to presenting a claim issued by a local security token service) providing a security Identity Provider Security Token Service (IP-STS) such as Active Directory Federation Services 2.0 (ADFS 2.0). Your services, whether on-premises or in the cloud, can then be configured to trust credentials, in the form of claims, issued by the IP-STS. Think of the IP-STS as simply the component that can tell if a username and password credentials are valid. In this approach, clients authenticate against the IP-STS; for example, by sending their Windows credentials to ADFS, and if valid, they receive claims they can subsequently present to your websites or services for access. Your websites or services only have to evaluate these claims when executing authorization logic - Windows Identity Foundation (WIF) provides these facilities.

For additional information, review this session - SIA305 Windows Identity Foundation and Windows Azure for Developers. In this session you can learn how Windows Identity Foundation can be used to secure your Web Roles hosted in Windows Azure, how you can take advantage of existing on-premises identities, and how to make the most of the features such as certificate management and staged environments. learn about how Windows Identity Foundation can be used to secure your Web Roles hosted in Windows Azure, how you can take advantage of existing on-premises identities, and how to make the best of features in our cloud offering, such as certificate management and staged environments learn about how Windows Identity Foundation can be used to secure your Web Roles hosted in Windows Azure, how you can take advantage of existing on-premises identities, and how to make the best of features in our cloud offering, such as certificate management and staged environments learn about how Windows Identity Foundation can be used to secure your Web Roles hosted in Windows Azure, how you can take advantage of existing on-premises identities, and how to make the best of features in our cloud offering, such as certificate management and staged environments

In the next release (version 2) of AppFabric Azure Access Control Service, a similar approach can be taken by using the Access Control Service (ACS v2) to act as your IP-STS and thereby replacing ADFS. With ACS, you can define service identities that are essentially logins secured by either a username and password, or a certificate. The client calls out to ACS first to authenticate, and the client thereafter presents the claims received by ACS in the call to your service or website.

Finally, more advanced scenarios can be put into place that uses ACS as your Relaying-Party Security Token Service (RP-STS) as the authority for all your apps, both on-premises and in the cloud. ACS is then configured to trust identities issued from other identity providers and convert them to the claims expected by your applications. For example, you can take this approach to enable clients to authenticate using Windows Live ID, Google, and Yahoo, while still working with the same set of claims you built prior to supporting those services.

Port Bridge, a point-to-point tunneling scenario, is perfect when the application is not exposed as a WCF service or doesn’t speak HTTP. Port bridge is perfect for protocols such as SMTP, SNMP, POP, IMAP, RDP, TDS and SSH. You can read more about this here.

Optimizing Bandwidth – Data transfer Costs

One huge issue to reconcile in hybrid solutions is the bandwidth – this is unique to hybrid.

Keeping Data Transfers In Your Datacenter

The amount of data transferred in and out of a datacenter is billable, so an approach to optimizing your bandwidth cost is to keep as much traffic from leaving the datacenter as possible. This means architecting designs that transfer most data across the Azure “fabric” within affinity groups, and minimizing traffic between disparate data centers and to external sites. An affinity group ensures that your cloud services and storage are hosted together on the Windows Azure infrastructure. Windows Azure roles, Windows Azure storage, and SQL Azure services can be configured at creation time to live within a specific datacenter. By ensuring the Azure hosted components within your hybrid solution that communicate with each other live in the same datacenter, you effectively eliminate the data transfer costs.

Caching Data, On-Premises And In The Cloud

The second approach to optimizing bandwidth costs is to use caching. This means leveraging Windows Server AppFabric Cache, an on-premises technology, to minimize calls to SQL Azure or Azure Storage. Similarly, it also means utilizing the Azure AppFabric Cache from Azure roles to minimize calls to SQL Azure, Azure Storage, or to on-premises services via Service Bus.

Optimizing Resources

One of the most significant cost-optimization features of Windows Azure roles is their support for dynamic scaling, up and down. This means adding more instances as your load increases and removing them as the load decreases. Supporting load-based dynamic scaling can be accomplished using the Windows Azure management API and SQL Azure storage. While there is no out-of-the box support for this, there is a fair amount of guidance on how to implement this within your solutions (see the Additional References section for one such example). The typical approach is to configure your Azure Web and Worker roles to log health metrics (e.g., CPU load, memory usage, etc.) to a table in SQL Azure. You then have another Worker role that is responsible for periodically polling this table and when certain thresholds are reached or exceeded, it would increase or decrease the instance count for the monitored service using the Windows Azure management APIs.

Monitoring & Diagnostics

For on-premises services, you can use Windows Server AppFabric Monitoring which logs tracking events into your on-premises instance of SQL Server, and this data can be viewed both through IIS Manager and by querying the monitoring database. For Azure roles in production you can leverage Windows Azure Diagnostics, wherein your instances periodically load batches to Azure storage. For SQL Azure you can examine SQL Azure query performance by querying the related Dynamic Management Views.

IntelliTrace is very useful for debugging solutions. With IntelliTrace debugging, you can log extensive debugging information for a Role instance while it is running in Windows Azure. If you need to track down a problem, you can then use the IntelliTrace logs to step through your code from within Visual Studio as though it were running in Windows Azure. In effect, IntelliTrace records key code execution and environment data while your service is running, and allows you to replay the recorded data from within Visual Studio. For more information on IntelliTrace, see Debugging with IntelliTrace.

Summary

Based off the discussions above, let’s consolidate our findings into this easy to read table.

Table 3: In Summary

Typical Scenario	Architectural Pattern	Application Host	Data Store(s)	Access Control	Management
EAI and Process Control	On Premise	IIS, Windows Server, BizTalk Server	AppFabric Cache, SQL Server	Active Directory, SQL Login	SQL Server Management Studio, IIS – Manager
Web content delivery	Cloud	Windows Azure – Web or Worker Roles	SQL Azure, Azure Storage, AppFabric Cache	Account name/key pair (limited IP Address access), SQL Login, Access Control Services	SQL Server Management Studio, Database Manager, Windows Azure Portal Management API.
Multi-location devices/ Appliance Management	Hybrid	IIS, Windows Server, BizTalk Server, Windows Azure – Web or Worker Roles Service Bus Connect	AppFabric Cache, SQL Server, SQL Azure, Azure Storage, AppFabric Cache	ADFS 2.0, Access Control Services	SQL Server Management Studio, Database Manager, Windows Azure Portal, IIS - Manager

It’s clear that the Hybrid Cloud Architectural pattern is ideally suited for connecting on-premises based appliances and applications with its cloud counterparts. This hybrid pattern is ideal since it leverages the best of both: traditional on-premises applications and new distributed/multi-tenant cloud applications.

Let’s Make Plans To Build This Out Now!

We have presented a survey of available technologies to build out the hybrid applications; other recent announcements at PDC 10 are worth considering before you move forward.

First, you should consider the Composition Model within the Windows Azure Composite Applications Environment. The Composition Model ties together your entire Azure hosted services (roles, storage, SQL Azure, etc.) as one composite application enabling you to see the end to end picture in a single designer in Visual Studio. The Composition Model is a set of .NET Framework extensions and builds on the familiar Azure Service Model concepts and adds new capabilities for describing the application infrastructure, its components, and relationships among the components, in order to treat the composite application as a single logical identity.

Second, if you were making calls back to an on-premises instance of Windows Server AppFabric, Windows Workflow is fully supported on Azure as part of the composition model, and so you may choose to move the relevant Workflows directly into the cloud. This is huge, very huge since it enables develop once and deploy as needed!

Third, there is a new role in town - the Windows Azure Compute VM Role. This enables us to host 3^rd party and archaic Line-of-business applications—and fully participate (wow - without a major code do-over!) in the hybrid cloud solution.

And last but not least, Windows Azure is also providing the Remote Desktop functionality, which enables you to connect to a running instance of your application or service to monitor activity and troubleshoot common problems. The Remote Desktop Protocol Access while super critical to the VM instance is also available to Web and Worker roles too. This in turn enables another related new feature, which is the availability of the full IIS instead of just the core that was available to web roles.

Additional References

While web links for additional/background information are embedded into the text; the following additional references are provided as resources to move forward on your journey.

Stay Tuned!

This is the beginning; our team is currently working on the next blog that builds on this topic. The blog will take a scenario driven approach in building a Hybrid Windows Azure Application.

Acknowledgements

Significant contributions from Paolo Salvatori, Valery Mizonov, Keith Bauer, Rama Ramani, Zoiner Tejada and Sidney Higa are acknowledged.

Namaste!

BizTalk Server–distinguished field

Quoc Bui — Tue, 30 Nov 2010 01:16:31 GMT

The distinguished field is used to expose the value of a node of a message instance to BizTalk Server system. However, if there is a record that has a parent node that has multiple instances, then that record won’t be able to be highlighted as distinguished. An error message of “The node can occur potentially multiple times in the instance. Only nodes which are guaranteed to be unique can be promoted.“ will appear.

This constraint was purposely designed to remove any ambiguity in accessing the data. However, despite that, should the developer still desire to use the node, then the record’s XPath query can still be used. The article Using XPaths in Message Assignments describes how XPath queries can be leveraged in BizTalk Server.

Technorati Tags: BizTalk Server,Distinguished,fields,XPath,queries

Windows Azure AppFabric Caching Service - soup to nuts primer

Rama Ramani — Mon, 29 Nov 2010 19:59:19 GMT

Introduction

One of the important reasons to move an application to the cloud is scalability (outside of other benefits that the cloud provides). When the application is deployed to the cloud, it is critical to maintain performance - the system needs to handle the increase in load and importantly maintain low latency response times. This is currently needed for most applications, be it a website selling books or a large social networking website or a complex map reduce algorithm. Distributed Caching as a technology enables applications to scale elastically while maintaining the application performance. Previously, Windows Server AppFabric enabled you to use distributed caching on-premises. Now, distributed caching is available on the cloud for Azure applications!

Here are some scenarios to leverage the Windows Azure AppFabric Caching Service:

ASP.NET application running as a web role needing a scalable session repository
Windows Azure hosted application frequently accessing reference data
Windows Azure hosted application aggregating objects from various services and data stores

Windows Azure AppFabric Caching is Microsoft's distributed caching Platform As A Service that can be easily configured and leveraged by your applications. It provides elastic scale, agile apps development, familiar programming model and best of all, managed & maintained by Microsoft.

At this time, this service is in labs release and can be used for development & test purposes only.

Configuring the Cache endpoint

Go to https://portal.appfabriclabs.com/ and login using your Live ID
Create a Project and then add a unique Service Namespace to it, for instance 'TestAzureCache1'.
When you click on the Cache endpoint, you should see the service URL and the authentication access token as follows:

At this point, you are done setting up the cache endpoint. Comparing this to installing and configuring the cache cluster on-premises shows the agility that cloud platforms truly provide.

Developing an application

Download the Windows Azure AppFabric SDK to get started.
Copy the app.config or web.config from the portal shown above and use it from your client application
In your solution, add references to Microsoft.ApplicationServer.Caching.Client and Microsoft.ApplicationServer.Caching.Core from C:\Program Files\Windows Azure AppFabric SDK\V2.0\Assemblies\Cache
Add the namespace to the beginning of your source code and develop the application in a similar manner to developing on-premises application

using Microsoft.ApplicationServer.Caching;

* At this point, Azure AppFabric caching service does not have all the features that Windows Server AppFabric Cache supports. . For more information, see http://msdn.microsoft.com/en-us/library/gg278356.aspx.

If you are developing the application via code instead of config, here are set of code for accessing the caching service. Only the statements highlighted need to be modified as per your configured endpoint.

private void PrepareClient()

{

// Insert the Service URL from the labs portal

string hostName = "TestAzureCache1.cache.appfabriclabs.com";

int cachePort = 22233;

List<DataCacheServerEndpoint> server = new List<DataCacheServerEndpoint>();

server.Add(new DataCacheServerEndpoint(hostName, cachePort));

DataCacheFactoryConfiguration config = new DataCacheFactoryConfiguration();

//Insert the Authentication Token from the labs portal

string authenticationToken = <To Be Filled Out>;

config.SecurityProperties = new DataCacheSecurity(authenticationToken);

config.Servers = server;

config.IsRouting = false;

config.RequestTimeout = new TimeSpan(0, 0, 45);

config.ChannelOpenTimeout = new TimeSpan(0, 0, 45);

config.MaxConnectionsToServer = 5;

config.TransportProperties = new DataCacheTransportProperties() { MaxBufferSize = 100000 };

config.LocalCacheProperties = new DataCacheLocalCacheProperties(10000, new TimeSpan(0, 5, 0), DataCacheLocalCacheInvalidationPolicy.TimeoutBased);

DataCacheFactory myCacheFactory = new DataCacheFactory(config);

myDefaultCache = myCacheFactory.GetCache("default");

}

Performance data

By now, you must be thinking if this is indeed this easy, what is the catch? :) :) There is none. Here are some performance numbers that I ran, hopefully this makes you feel at ease.

This is an ASP.NET application running as a web role in Windows Azure accessing the cache service. The Object count parameter is used to instantiate an object which has an array of int[] and string[]. For example, when Object count is set to 100 the application will instantiate int[100] and string[100] with each array item is initialized to a string ~6 bytes. So approximately, the parameter 'Object count' * 10 is the pre-serialized object size. When storing the object in the caching service, the KEY is generated using RAND and then depending on the Operations dropdown, the workload is run for a set of iterations based on the 'Iterations' parameter. Finally, the average latency is computed. The web role is running in South Central US data center in the same DC where the caching service is also deployed in order to avoid any external network latencies. For all these tests, the local cache feature is disabled.

Operation	Object Size (pre-serialized)	Avg Latency (ms)
PUT	~1KB	4.5 - 6.5
GET	~ 1 KB	4.8 - 5.1
PUT	~10 KB	7.4 - 8.9
GET	~10 KB	7.6 - 10.3
PUT	~100 KB	38.9 - 45.6
GET	~100 KB	4.6 - 7.3

Now that you have some cache in the cloud, get your apps going!

Authored by: Rama Ramani
Reviewed by: Christian Martinez, Jason Roth

StreamInsight: Understanding dynamic query composition

Mark Simms [MSFT] — Mon, 22 Nov 2010 23:40:00 GMT

Been tied up with PASS for the past (pun intended) couple of weeks, so it’s time to get a bit caught up on writing. One of the key technical features of StreamInsight is the ability for one query to consume the output of another, enabling the system to avoid having to process events twice, and opening up a new world of flexibility. The feature that unlocks these capabilities is Dynamic Query Composition. That being said, some of the nuances of how StreamInsight constructs queries and flows events is not always obvious. As this has resulted in me making a few sub-optimal design choices in the past , I figured I’d put together a little article demonstrating the subtle nuances involved when using (or not using) DQC.

For more background on DQC, please refer to:

Composing Queries at Runtime

Take the diagram below, illustrating a common query composition pattern. Given a single input stream (represented by the inputAdapter shape in the diagram), we would like to consume the stream of events from that adapter in both queryOne and queryTwo.

At first glance, this query syntax would seem to fit the bill (note that I’m using the new IObservable support for the input adapter, but using the classic adapter syntax to create the queries bound to an output adapter, to explicitly show query creation):

Code Snippet
 // Convert the data source into a temporal stream
var orderStream = ordersSimple.ToPointStream(cepApp, s =>
    PointEvent.CreateInsert(s.StartTime, s),
    AdvanceTimeSettings.IncreasingStartTime);
 
// Create a query in two parts
var queryOne = from e in orderStream where e.OrderID > 50 select e;
var queryTwo = from e in queryOne.TumblingWindow(
    TimeSpan.FromDays(1), HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
    OrderCount = e.Count()
};
 
// Convert these templates into queries, bound to an output adapter
var tracerConfig = new TracerConfig()
{
        DisplayCtiEvents = false,
        SingleLine = true,
        TracerKind = TracerKind.Console
};
 
var queryOneRun = queryOne.ToQuery(cepApp, "QueryOne", "",
    typeof(TracerFactory), tracerConfig, EventShape.Point,
    StreamEventOrder.FullyOrdered);
 
var queryTwoRun = queryTwo.ToQuery(cepApp, "QueryTwo", "",
    typeof(TracerFactory), tracerConfig, EventShape.Interval,
    StreamEventOrder.FullyOrdered);
 

However, this wouldn’t produce the desired result. Instead, the StreamInsight engine will construct the query pattern seen in the diagram below – where queryTwo composes the design of queryOne, but not the runtime stream (i.e. creates two adapter instances).

This can be verified by using the debugger to check the input source of the query. For details on how to use the debugger, see my blog post here.

Open the Event Flow Debugger, and connect to your StreamInsight instance (walkthrough on using the Event Flow Debugger here).
Navigate to your queryTwo definition, and Show Query.
As seen in the diagram below, if the input adapter is an actual input adapter (and not a published stream) you have two independent streams each with their own input adapter.

In order to have the two queries compose at runtime, we need to:

Publish the results of the first query as a published stream with a strongly typed output (published schemas cannot have anonymous types)
Consume the published stream as the input source of the second query.

Updating the query syntax to take advantage of DQC looks like:

Code Snippet
 // Create a simple data source (oData feed), using a non-anonymous
// type
var ordersSimple = from o in northwind.Orders
                    where o.OrderDate != null && o.ShippedDate != null
                    select new NorthwindOrderResult
                    {
                        StartTime = (DateTime)o.OrderDate,
                        EndTime = (DateTime)o.ShippedDate,
                        OrderID = o.OrderID,
                        ShipRegion = o.ShipRegion,
                        CompanyName = o.Customer.CompanyName
                    };
 
// Convert the data source into a temporal stream
var orderStream = ordersSimple.ToPointStream(cepApp, s =>
    PointEvent.CreateInsert(s.StartTime, s),
    AdvanceTimeSettings.IncreasingStartTime);
 
// Create a query 
var queryOne = from e in orderStream where e.OrderID > 50 select e;                     
var queryOneRun = queryOne.ToQuery(cepApp, "QueryOne", "",
    typeof(TracerFactory), tracerConfig, EventShape.Point,
    StreamEventOrder.FullyOrdered);
 
// Convert the query's output into a published stream (with a 
// non-anonymous type)
var queryOneStream = queryOneRun.ToStream<NorthwindOrderResult>();
 
// Bind the second query to the published stream
var queryTwo = from e in queryOneStream.TumblingWindow(
    TimeSpan.FromDays(1), HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
    OrderCount = e.Count()
};
           
// Run the second query
var queryTwoRun = queryTwo.ToQuery(cepApp, "QueryTwo", "",
    typeof(TracerFactory), tracerConfig, EventShape.Interval,
    StreamEventOrder.FullyOrdered);
 

Now that we’ve updated the query structures to use DQC, let’s take another look at the query structure in the debugger. Note that the input source for the second query is now a published stream (as it is now bound to a URI source, not a direct input adapter).

There we go – we can now feed the run-time results of one query into another query using Dynamic Query Composition.

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

Paolo Salvatori — Mon, 22 Nov 2010 23:35:00 GMT

Introduction

Some time ago, I was asked by a customer whether or not BizTalk WCF Adapters support the Duplex Message Exchange Pattern. The latter 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. This form of bi-directional communication is useful for services that must communicate directly to the client or for providing an asynchronous experience to either side of a message exchange, including event-like behavior. The BizTalk WCF Adapters support this message exchange pattern, but unfortunately this feature is undocumented. Therefore, I decided to create a full demo where a client .NET application exchange messages with an Orchestration via a two-way Request-Response WCF Receive Location that supports callbacks and duplex communication. Subsequently, I decided to extend this demo and introduce a WCF Workflow Service between the WinForm client application and the underlying Orchestration to implement a more complex scenario The final objective was to investigate how exploiting the Correlation mechanisms provided by WF 4.0 to get a workflow running within IIS/AppFabric to exchange messages with a downstream orchestration in an asynchronous mode.

This article will be composed of 3 parts:

Part 1: How to exchange messages with an orchestration via a two-way WCF Receive Location using the Duplex Message Exchange Pattern.
Part 2: 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.
Part 3: How to implement an asynchronous communication between a WCF Workflow Service and an Orchestration using WS-Addressing and Content-Based Correlation.

Why Duplex Message Exchange Pattern?

The Duplex Message Exchange Pattern is fully supported by WCF and is extensively documented on MSDN. As I already mentioned in the introduction, the Duplex pattern allows two applications to act both as service endpoints and send messages to each other independently in an asynchronous way. There are many reasons for using the Duplex Message Exchange Pattern when communicating with a BizTalk solution:

If an orchestration invoked by the consumer application takes seconds to minutes to complete.
If you are invoking an orchestration within an ASP.NET page, use asynchronous pages.
If you are invoking an orchestration from a single threaded application, such as a Windows Forms or Windows Presentation Foundation (WPF) application. When using the event-based asynchronous calling model, the result event is raised on the UI thread, adding responsiveness to the application without requiring you to handle multiple threads yourself.

In general, if you have a choice between a synchronous and asynchronous call, you should choose the latter approach. In fact, a synchronous call typically blocks the client thread till the operation completes, whereas an asynchronous call is non-blocking and only initiates the operation. This way the consumer application can continue its execution without waiting for the call to complete and get notified by the service when the result of its request is ready.

For more information on how implementing the Duplex Message Exchange Pattern with WCF, you can read the following articles:

"Synchronous and Asynchronous Operations" topic on MSDN.
"Designing Service Contracts" topic on MSDN.
“How to: Create a Duplex Contract” topic on MSDN.
“Duplex Services” topic on MSDN.
“Choosing a Message Exchange Pattern” topic on MSDN.

Using a Duplex Channel to communicate with a BizTalk Application

The following picture represents the architecture of the first use case. The idea behind the application is quite straightforward: a Windows Forms application submits a question to an orchestration called SyncMagic8Ball 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.

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 Two-Way WCF-NetTcp Receive Location.
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 finally returned to the original caller.

WCF Receive Locations

In order to establish an asynchronous communication, the client application and service, in our case BizTalk Server, have to meet the following conditions:

They need to be online simultaneously.
They need to use the same WCF binding.
The selected binding must support sessions and duplex contracts. WCF provides a rich set of built-in bindings that support the Duplex MEP. For more information, you can read “System-Provided Bindings” topic on MSDN.
The client application needs to implement a Service Contract to submit the request message to BizTalk and expose a Callback Contract to receive the response message from BizTalk Server asynchronously.

To satisfy the second and third conditions, I have created 3 different Two-Way Request-Response Receive Locations with the following characteristics:

The 3 Receive Locations are children of the same Receive Port called DuplexMEP.Sync.ReceivePort.
They use the XmlReceive pipeline to process the inbound message and the PassThruTrasmit pipeline to process the outbound message.
The DuplexMEP.Sync.WCF-NetTcp.ReceiveLocation uses the WCF-NetTcp Adapter and is hosted by the BizTalkServerReceiveHost64 in-process host.
The DuplexMEP.Sync.WCF-NetNamedPipe.ReceiveLocation uses the WCF-NetNamedPipe Adapter and is hosted by the BizTalkServerReceiveHost64 in-process host.
The DuplexMEP.Sync.WCF-Custom.WsDualHttpBinding.ReceiveLocation uses the WCF-CustomIsolated Adapter in conjunction with the WsDualHttpBinding and is hosted by the BizTalkServerIsolatedHost.

The following picture shows the 3 Receive Locations inside the BizTalk Administration Console:

Note
You should select the most appropriate WCF Adapter and binding based on your needs:

The WsDualHttpBinding binding is designed for interoperability with advanced web services that need to support WS-* standards. It uses the HTTP protocol for the transport and the text message encoding.
NetTcpBinding and NetNamedPipeBinding, on the other hand, are designed for efficient and performant communication with other WCF applications across machines or on the same machine respectively. They both use the binary message encoding that guarantees better performance than the text message encoding.

Bindings have different characteristics in terms of response time and throughput, so the general advice to increase performance is using the NetTcpBinding and NetNamedPipeBinding whenever possible.

Schemas

The request messages sent by the Windows Forms Client Application have the following format:

Request Message

<Request xmlns="http://microsoft.appfabric.cat/10/samples/duplexmep">
    <Id>9aff8596-ec87-494d-801e-30f286d449a4</Id>
    <Question>Will the world end in 2012?</Question>
    <Delay>0</Delay>
</Request>

The corresponding response messages returned by the SyncMagic8Ball orchestration have the following format:

Response Message

<Response xmlns="http://microsoft.appfabric.cat/10/samples/duplexmep">
    <Id>9aff8596-ec87-494d-801e-30f286d449a4</Id>
    <Answer>Most likely</Answer>
</Response>

Both message types are defined by an XML Schema contained in the Schemas project that you can find in the companion code for this article.

Orchestration

The following picture shows the structure of the SyncMagic8Ball orchestration.

As you can easily notice, the orchestration uses the Two-Way Request-Response Logical Port to receive the inbound request message and 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.

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.

Note
To extract the value of individual fields from the inbound document, I could have used Distinguished Fields defined on the XML Schema for the request message. These are context properties whose name is defined by the XPath Expression used to retrieve data from an input XML document. If the XML Schema of the document is quite complex, the XPath Expression in question can be very long and therefore the corresponding Distinguished Field can occupy a significant amount of space in the message context. Therefore, I recommend to implement the following best practices in any BizTalk application:

Eliminate unnecessary and unused Distinguished Fields.
Use a different technique to access the value of individual elements and attributes within an Xml document.

When the document in question is extremely small, you can also load the message into an XmlDocument object and use the SelectSingleNode method to extract the information you need using an XPath Expression. However, the general advice is to minimize the usage of XmlDocument variables in orchestrations and in .NET code. Loading a message into an XmlDocument variable has significant overhead, especially for large messages. This overhead is in terms of memory usage and system resources required to build the in-memory structures. The use of an XmlDocument instance forces the message content to be entirely loaded into memory in order to build the object graph for the DOM. The total amount of memory used by an instance of this class can be around 10 times the actual message size. For more information and evidence, see the following articles I wrote on this topic:

”4 Different ways to process an XLANGMessage within an helper component invoked by an orchestration Part 1” on the AppFabric CAT blog.
”4 Different ways to process an XLANGMessage within an helper component invoked by an orchestration Part 2” on the AppFabric CAT blog.

The request message in my demo is relatively small; nevertheless, I decided to use an helper class that adopts a streaming approach to extract data from the XLANGMessage passed in as argument:

public class XPathHelper
{
    #region Private Constants 
    private const string MessageCannotBeNull = "[XPathReader] The message cannot be null.";
    #endregion
        
    #region Public Static Methods
    public static string GetValue(XLANGMessage message, int partIndex, string xpath)
    {
        try
        {
            if (message == null)
            {
                throw new ApplicationException(MessageCannotBeNull);
            }
            using (Stream stream = message[partIndex].RetrieveAs(typeof(Stream)) as Stream)
            {
                XmlTextReader xmlTextReader = new XmlTextReader(stream);
                XPathCollection xPathCollection = new XPathCollection();
                XPathReader xPathReader = new XPathReader(xmlTextReader, xPathCollection);
                xPathCollection.Add(xpath);
                while (xPathReader.Read())
                {
                    if (xPathReader.HasAttributes)
                    {
                        for (int i = 0; i < xPathReader.AttributeCount; i++)
                        {
                            xPathReader.MoveToAttribute(i);
                            if (xPathReader.Match(xPathCollection[0]))
                            {
                                return xPathReader.GetAttribute(i);
                            }
                        }
                    }
                    if (xPathReader.Match(xPathCollection[0]))
                    {
                        return xPathReader.ReadString();
                    }
                }
            }
        }
        finally
        {
            message.Dispose();
        }
        return string.Empty;
    }
    #endregion
}

The Set Response Message Assignment shape invokes the SetResponse static method exposed by the ResponseHelper class to generate the response message:

ResponseHelper Class

public class ResponseHelper
{
    #region Private Constants
    private const string MessageFormat = "[ResponseManager] {0}";
    private const string ResponseNamespace = "http://microsoft.appfabric.cat/10/samples/duplexmep";
    private const string Response = "Response";
    private const string Id = "Id";
    private const string Answer = "Answer";
    #endregion

    #region Static Constructor
    private static string[] answers = null;
    #endregion

    #region Static Constructor
    static ResponseHelper()
    {
        answers = new string[]{"As I see it, yes",
                                "It is certain",
                                "It is decidedly so",
                                "Most likely",
                                "Outlook good",
                                "Signs point to yes",
                                "Without a doubt",
                                "Yes",
                                "Yes – definitely",
                                "You may rely on it",
                                "Reply hazy, try again",
                                "Ask again later",
                                "Better not tell you now",
                                "Cannot predict now",
                                "Concentrate and ask again",
                                "Don't count on it",
                                "My reply is no",
                                "My sources say no",
                                "Outlook not so good",
                                "Very doubtful"};
    }
    #endregion

    #region Public Static Methods
    public static void SetResponse(XLANGMessage message, string id)
    {
        try
        {
            VirtualStream stream = new VirtualStream();
            using (XmlWriter writer = XmlWriter.Create(stream))
            {
                writer.WriteStartDocument();
                writer.WriteStartElement(Response, ResponseNamespace);
                writer.WriteStartElement(Id, ResponseNamespace);
                writer.WriteString(id);
                writer.WriteEndElement();
                writer.WriteStartElement(Answer, ResponseNamespace);
                Random random = new Random(unchecked((int)DateTime.Now.Ticks));
                writer.WriteString(answers[random.Next(0, 20)]);
                writer.WriteEndElement();
                writer.WriteEndElement();
            }
            stream.Seek(0, SeekOrigin.Begin);
            message[0].LoadFrom(stream);
        }
        catch (Exception ex)
        {
            LogHelper.WriteLine(ex.Message);
        }
        finally
        {
            message.Dispose();
        }
    }
    #endregion
}

As you can see, the ResponseHelper class uses a VirtualStream object an XmlWriter class instance to initialize the content of the response message. Once again, the response message in this case is so small that I could have used an XmlDocument object to load the content of the response message. However, my intention was to show you how to take advantage of the streaming approach to read and write the content of an XLANGMessage object within an orchestration.

Let’s take a look at the code used by the Windows Forms Client Application to send a request message and receive the corresponding response using a Callback contract.

Data and Message Contracts

I started defining the Data and Message Contracts for the request and response messages. These 2 types of contracts have different roles in WCF:

Data Contracts provide a mechanism to map .NET CLR types that are defined in code and XML Schemas (XSD) defined by the W3C organization (www.w3c.org). Data contracts are published in the service’s metadata, allowing clients to convert the neutral, technology-agnostic representation of the data types to their native representations.
Message Contracts describe the structure of SOAP messages sent to and from a service and enable you to inspect and control most of the details in the SOAP header and body. Whereas data contracts enable interoperability through the XML Schema Definition (XSD) standard, message contracts enable you to interoperate with any system that communicates through SOAP. Using message contracts gives you complete control over the SOAP message sent to and from a service by providing access to the SOAP headers and bodies directly. This allows the use of simple or complex types to define the exact content of the SOAP parts.

For more information on Data and Message Contracts, you can read the following articles:

"Using Data Contracts" topic on MSDN.
“Using Message Contracts” topic on MSDN.

In my solution, I created 2 separate projects called DataContracts and MessageContracts respectively. For your convenience, I included below the code of the 4 classes used to define the Data and Message Contract of the request and response messages.

BizTalkRequest Class

[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 Class

[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
}

BizTalkRequestMessage Class

[MessageContract(IsWrapped=false)]
public class BizTalkRequestMessage
{
    #region Private Fields
    private BizTalkRequest request;
    #endregion

    #region Public Constructors
    public BizTalkRequestMessage()
    {
        this.request = null;
    }

    public BizTalkRequestMessage(BizTalkRequest request)
    {
        this.request = request;
    }
    #endregion

    #region Public Properties
    [MessageBodyMember(Namespace = "http://microsoft.appfabric.cat/10/samples/duplexmep")]
    public BizTalkRequest Request
    {
        get
        {
            return this.request;
        }
        set
        {
            this.request = value;
        }
    } 
    #endregion
}

BizTalkResponseMessage Class

[MessageContract(IsWrapped = false)]
public class BizTalkResponseMessage
{
    #region Private Fields
    private BizTalkResponse response;
    #endregion

    #region Public Constructors
    public BizTalkResponseMessage()
    {
        this.response = null;
    }

    public BizTalkResponseMessage(BizTalkResponse response)
    {
        this.response = response;
    }
    #endregion

    #region Public Properties
    [MessageBodyMember(Namespace = "http://microsoft.appfabric.cat/10/samples/duplexmep")]
    public BizTalkResponse Response
    {
        get
        {
            return this.response;
        }
        set
        {
            this.response = value;
        }
    }
    #endregion
}

Indeed, I could have used just Data Contracts to model messages as Message Contracts add a degree of complexity. So why I decided to use Message Contracts? Well, the reason is quite straightforward. By assigning false to the IsWrapped property exposed by the MessageContractAttribute, you specify that the message body won’t be contained in a wrapper element. Typically, the wrapper element of a request message is the name of the operation invoked and it’s defined in the WSDL. Setting the value of the IsWrapped property to false, you can simply select the Body option in both the Inbound BizTalk message body and Outbound WCF message body sections on the Messages tab when configuring a WCF Receive Location. On the contrary, you should define a Path in the Inbound BizTalk message body section to extract the payload from the inbound message, and specify a template in the Outbound WCF message body section to include the outgoing response message within a wrapper element.

Note
The namespace of the Message and Data Contract classes match those defined by the XML Schemas that model the request and response messages in the BizTalk Server application.

Service Contracts

The next step was to define the Service Contracts used by the client application to exchange messages with BizTalk Server. To this purpose, I created a new project in my solution called ServiceContracts, and then I started off by defining the Service Contract Interface that models the server side of the duplex contract. I declared the signature of a One-Way method called AskQuestion. I specified a parameter of type BizTalkRequestMessage (see the Data and Message Contracts section) and void as return type. Then I decorated the method with the XmlSerializerFormatAttribute to indicate to the WCF Runtime to use the XmlSerializer instead of the DataContractSerializer. Please note that the WCF Adapters and BizTalk in general use the XmlSerializer, thus I had to explicitly set the correct serializer in the contract definition. Next, I decorated the method with the OperationContractAttribute to indicate that the method is one way and to specify the WS-Addressing Action of the request message. Finally I decorated the interface with the ServiceContractAttribute.

IMagic8BallBizTalk Interface

[ServiceContract(Namespace = http://microsoft.appfabric.cat/10/samples/duplexmep,
                 SessionMode = SessionMode.Required,
                 CallbackContract = typeof(IMagic8BallBizTalkCallback),
                 ConfigurationName = "IMagic8BallBizTalk")]
public interface IMagic8BallBizTalk
{
    [XmlSerializerFormat]
    [OperationContract(Action = "AskQuestion", IsOneWay = true)]
    void AskQuestion(BizTalkRequestMessage requestMessage);
}

Then I defined the the callback interface as follows:

IMagic8BallBizTalkCallback Interface

[ServiceContract(Namespace = http://microsoft.appfabric.cat/10/samples/duplexmep, 
                 ConfigurationName = "IMagic8BallBizTalkCallback")]
public interface IMagic8BallBizTalkCallback
{
    [XmlSerializerFormat]
    [OperationContract(Action = "AskQuestionResponse", IsOneWay = true)]
    void AskQuestionResponse(BizTalkResponseMessage responseMessage);
}

Finally, I linked the two interfaces into a Duplex Contract by assigning the type of the callback interface to the CallbackContract property in the ServiceContractAttribute that decorates the IMagic8BallBizTalk interface.

Callback Contract Implementation

The next step was to implement the IMagic8BallBizTalkCallback callback interface in order to receive response messages from BizTalk Server. To this purpose I created a new class in the Client project called Magic8BallBizTalkCallback that implements the IMagic8BallBizTalkCallback callback interface.

Magic8BallBizTalkCallback Class

[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
}

...

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

In particular, when the WCF Receive Location invokes the AskQuestionResponse method to return a response message, the callback is handled by a ThreadPool thread other than the main thread running the client application. Since the InternalWriteToLog method is used to write the answer on a ListBox control owned by the main thread, the WriteToLog method exposed by the MainForm class uses the InvokeRequired property to check whether it's necessary to use the Invoke method because the caller is on a different thread than the main thread.

Invoking BizTalk Server

The next step was to write the code to invoke one of the 3 WCF Receive Locations exposed by by BizTalk application.

btnAsk_Click Method

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

        if (string.IsNullOrEmpty(cboEndpoint.Text))
        {
            WriteToLog(NoEndpointsFound);
        }

        DuplexChannelFactory<IMagic8BallBizTalk> channelFactory = null;

        try
        {
            BizTalkRequest request = new BizTalkRequest(txtQuestion.Text, delay);
            BizTalkRequestMessage requestMessage = new BizTalkRequestMessage(request);
            WriteToLog(string.Format(CultureInfo.CurrentCulture, 
                                     RequestFormat, 
                                     cboEndpoint.Text, 
                                     request.Id, 
                                     request.Question));
            InstanceContext context = new InstanceContext(new Magic8BallBizTalkCallback(this));
            channelFactory = new DuplexChannelFactory<IMagic8BallBizTalk>(context, cboEndpoint.Text);
            IMagic8BallBizTalk channel = channelFactory.CreateChannel();
            channel.AskQuestion(requestMessage);
        }
        catch (FaultException ex)
        {
            WriteToLog(ex.Message);
            if (channelFactory != null)
            {
                channelFactory.Abort();
            }
        }
        catch (CommunicationException ex)
        {
            WriteToLog(ex.Message);
            if (channelFactory != null)
            {
                channelFactory.Abort();
            }
        }
        catch (TimeoutException ex)
        {
            WriteToLog(ex.Message);
            if (channelFactory != null)
            {
                channelFactory.Abort();
            }
        }
        catch (Exception ex)
        {
            WriteToLog(ex.Message);
            if (channelFactory != null)
            {
                channelFactory.Abort();
            }
        }
    }
    catch (Exception ex)
    {
        WriteToLog(ex.Message);
    }
}

As you may have noticed, the method btnAsk_Click performs the following actions in order:

Creates a data contract object (BizTalkRequest)
Creates a message contract object (BizTalkRequestMessage).
Creates a new instance of the Magic8BallBizTalkCallback type to handle callbacks.
Creates a new InstanceContext object. In particular, the Magic8BallBizTalkCallback object is passed in as argument to the constructor of the InstanceContext object.
Instantiates a DuplexChannelFactory to create a duplex channel of type IMagic8BallBizTalk. In particular, the first argument specifies the InstanceContext that the client uses to listen for messages from the connected service, in our case BizTalk Server.
Creates a new duplex channel.
Invokes the AskQuestion method.

Configuration File

The final step was to define the client endpoints in the client configuration file in order to invoke the 3 WCF Receive Locations exposed by the BizTalk application.

Configuration File

<?xml version="1.0"?>
<configuration>
  <system.serviceModel>
    <bindings>
      <netNamedPipeBinding>
        <binding name="netNamedPipeBinding" />
      </netNamedPipeBinding>
      <netTcpBinding>
        <binding name="netTcpBinding">
          <security mode="Transport">
            <transport protectionLevel="None" />
          </security>
        </binding>
      </netTcpBinding>
      <wsDualHttpBinding>
        <binding name="wsDualHttpBinding" />
      </wsDualHttpBinding>
    </bindings>
    <client>
      <clear />
      <endpoint address="net.tcp://localhost:7171/magic8ballbiztalk/sync"
                binding="netTcpBinding" 
                bindingConfiguration="netTcpBinding"
                contract="IMagic8BallBizTalk"
                name="NetTcpEndpointBizTalk" />
      <endpoint address="net.pipe://localhost/magic8ballbiztalk/sync" 
                binding="netNamedPipeBinding"
                bindingConfiguration="netNamedPipeBinding" 
                contract="IMagic8BallBizTalk"
                name="NetNamedPipeEndpointBizTalk" />
      <endpoint address="http://localhost/magic8ballbiztalk/syncmagic8ball.svc"
                binding="wsDualHttpBinding" 
                bindingConfiguration="wsDualHttpBinding"
                contract="IMagic8BallBizTalk"
                name="WsDualHttpEndpointBizTalk" />
    </client>
  </system.serviceModel>
  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.0"/>
  </startup>
</configuration>

We are now ready to test the solution.

Testing the Solution

To test the application, you can proceed as follows:

Configure and 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 the following endpoints:
- NetTcpEndpointBizTalk to invoke the DuplexMEP.Sync.WCF-NetTcp.ReceiveLocation.
- NetNamedPipeEndpointBizTalk to invoke the DuplexMEP.Sync.WCF-NetNamedPipe.ReceiveLocation.
- WsDualHttpEndpointBiztalk to invoke the DuplexMEP.Sync.WCF-Custom.WsDualHttpBinding.ReceiveLocation.
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 BizTalk in an asynchronous way. Therefore, the client application doesn't need to wait for the response to the previous question before posing a new one.

In order for the demo to work, it’s important that the Delay in seconds that you specify in the client application is less than the ReceiveTimeout configured in the WCF Receive Location. When using a WCF Adapter other than the WCF-Custom and WCF-CustomIsolated, you cannot specify a custom ReceiveTimeout. The following picture shows the Binding tab of a WCF-NetTcp Receive Location.

As a consequence, the default 10 minutes timeout will be used. Therefore, if you need to increase the ReceiveTimeout beyond 10 minutes, you can replace the WCF Adapter in question with the WCF-Custom and WCF-CustomIsolated adapter and select the same binding. This way, you can specify a custom value for the ReceiveTimeout property exposed by the binding, as shown in the following picture:

Conclusions

In this article we have seen how to exchange messages with an orchestration via a two-way WCF Receive Location using the Duplex Message Exchange Pattern. In the next article of the series, we’ll see how to implement an asynchronous communication between our client application and a WCF Workflow Service running within IIS\AppFabric Hosting Services using the Durable Duplex Correlation. In the meantime, here you can download the companion code for this article. As always, you feedbacks are more than welcome!

Acknowledge review and comments from Valery Mizonov and Christian Martinez. Thanks guys!

A Configurable AppFabric Cache Attribute For Your WCF Services

Christian.Martinez — Wed, 10 Nov 2010 21:26:00 GMT

Doing a search online for WCF and Cache will eventually lead you to a number of links* demonstrating the use of an IOperationInvoker implementation to manage caching. With AppFabric Cache rapidly becoming a mainstream enterprise technology I felt this to be an ideal time to revisit this concept.

I chose to offer the following features in my implementation

Using the default or a named cache
Using a cache region
Action keyed caching
Choice of using one or more parameters to the operation as a key
Choice of XML serialized keys or a more compact form
Choice of XmlSerializer or DataContractSerializer when using the XML serialization option

Using an IOperationInvoker implementation allows for the following pattern

Using the custom invoker in your code is straightforward and natural as the following snippet from the test service provided with the code attached to this blog demonstrates

         /// <summary>
        /// This method is for testing cache retrieval by key.
        ///
        /// </summary>
        /// <param name="catGuy"></param>
        /// <returns></returns>
        [OperationContract]
        [AppFabricCacheBehavior]
        CatMember GetCat(long catGuy);

        /// <summary>
        /// This method is for testing demonstrating cache by Action.
        /// </summary>
        /// <returns></returns>
        [OperationContract]
        [AppFabricCacheBehavior(CacheByAction=true)]
        List<long> GetIds();

As you can see it’s only a matter of supplying an additional attribute along with your OperationContract..

The behavior code itself is very simple and contains descriptions of what the attribute parameters do

    [AttributeUsage(AttributeTargets.Method)]
    public class AppFabricCacheBehavior : Attribute, IOperationBehavior
    {
        /// <summary>
        /// If this property is true the only key considered is Action.
        /// A common use case scenario is a method that returns a slowly changing drop down list.
        /// Setting this will cause all other non AppFabric Cache specific options to be ignored.
        /// </summary>
        public bool CacheByAction { get; set; }

        /// <summary>
        /// If this property is true then either DataContractSerializer or XmlSerializer will be used on keys and then concatenated.
        /// By default this is false and the key will be a simple string concatenation 
        /// You should set this to true when using complex type(s) as a key.
        /// Avoid setting this to true when you are using primitive type(s) as a key.
        /// </summary>
        public bool SerializeKeysToXml { get; set; }

        /// <summary>
        /// Substitutes XmlSerializer for DataContractSerializer when SerializeKeysToXml is true.
        /// </summary>
        public bool UseXmlSerializer { get; set; }

        /// <summary>
        /// Retrieves items from the named cache if not null. If CacheName is not specified DefaultCache will be used. 
        /// </summary>
        public string CacheName { get; set; }

        /// <summary>
        /// Retrieves items from the named region.  
        /// </summary>
        public string CacheRegion { get; set; }

        /// <summary>
        /// Indexes of the objects to generate key from. If not specified it will default to KeyIndexes = new int[]{0}.  
        /// </summary>
        public int[] KeyIndexes { get; set; }

        public void AddBindingParameters(OperationDescription operationDescription, BindingParameterCollection bindingParameters)
        {
            
        }

        public void ApplyClientBehavior(OperationDescription operationDescription, ClientOperation clientOperation)
        {
            
        }

        public void ApplyDispatchBehavior(OperationDescription operationDescription, DispatchOperation dispatchOperation)
        {
            dispatchOperation.Invoker = new AppFabricCachingInvoker(this, dispatchOperation.Invoker,dispatchOperation.Action);
        }

        public void Validate(OperationDescription operationDescription)
        {
            
        }
    }

The only real work being done in this class is forwarding the original Invoker and passing in the Action.

Attribute Parameter Use Cases and Observations

While I was researching this article I noticed in some implementations they would unconditionally serialize all the parameters to XML and then concatenate them into a giant key. I still allow you to do this in my implementation if you choose to but I also give you a choice to use a simpler more compact format that will greatly reduce the memory usage when using primitive types. In simple test cases this cuts memory usage in 1/2 and I suspect in cases with very large payloads the effect is even more dramatic.

I also noticed some of the implementations did not take into account methods using zero parameters in their key calculations. It is a common occurrence to see Web Services built with zero parameter methods in order to provide for retrieving lists and such. By setting CacheByAction=true in the supplied code you will bypass key calculation logic and immediately use the Action parameter passed in as a key instead.

Test runs on my system ran approximately 3 times faster for cached responses versus having to access the database. As with any performance testing however, you will need to test in your own environment to see the effect a solution like the one presented might have.

Try It Out

Please give the code a test run and drop a comment if you would like a new feature or find what you think is a bug.

*The following links all demonstrate a form of caching using an IOperationInvoker implementation

Windows Communication Foundation Extensibility

Extending WCF with Custom Behaviors by Aaron Skonnard

Developing Custom OperationInvoker by Sankarsan

Use AppFabric Cache to cache your WCF Service response by Mikael Håkansson

Thanks to Paolo Salvatori, Valery Mizonov, and Quoc Bui for their review

Built for Quoc