softlogger Latest Articles ::Windows-Workflow-Foundation

Asynchronous Workflow Activities: Implemented Asynchronous WebRequests

Kirk Allen Evans Blog — 2007-07-02T00:00:00

(note: sample code is attached to this post).

As I stated before, Darma's excellent book finally gave me the a-ha moment for Windows Workflow Foundation beyond building simple Hello, World demos and making activities pretty. I finally get how to do more useful things with WF, and finally see why you should think twice before writing another program that doesn't use WF (not saying that every program requires WF, not by any stretch… but many of the programs that exist in the wild could seriously benefit from WF, enough that you should take a second look).

A task that I have wondered about is how you would perform work asynchronously. For instance, you want to utilize the IAsyncResult pattern when possible for server side applications, but how would you achieve that? Turns out that Windows Workflow Foundation makes this very straightforward. To demonstrate this, I am going to pirate some code from MSDN and repurpose it into an Activity. We will pirate some of the code from the MSDN documentation on using FtpWebRequest.

I posted to an internal discussion list asking how to do this, and Jon Flanders pointed out that he had written a sample that used the pattern I needed to follow. Since I hadn't yet grokked services, queues, and the real purpose of workflow as a runtime that spans multiple process and appdomain boundaries, I looked at it, and my eyes glazed over. Once I went back and took another peek, it made perfect sense. I think this is the single most important thing to grok for Windows Workflow Foundation: the concept of long-running workflow activities.

I am going to follow the same pattern Jon used, and we will pirate code from MSDN to fit within that pattern.

The really cool thing about this is that you will see how to take existing code and apply it to workflow activities, gaining new capabilities for your existing code. Developers will be able to take your activities and simply drag-and-drop their way through some fairly complex processing, becoming incredibly productive while you get to focus on the really cool plumbing underneath. You get to focus on writing reusable activities that developers can use incredibly quickly in their applications, and you get to focus on the more interesting deeper code. You get all the kudos for being the rock star who saves the company money by increasing developer productivity!

As with any WF Activity, we start by creating a class that will represent our Activity. Our activity will simply get an FTP file, we'll call it GetFileActivity.

using System;
using System.Net;
using System.Workflow.Runtime;
using System.Workflow.Activities;
using System.Workflow.ComponentModel;
using System.ComponentModel;

namespace Workflow.Ftp

{

public partial class GetFileActivity: SequenceActivity

{

public GetFileActivity()

{

InitializeComponent();

}

We will create two properties for our Activity, one that allows the user to specify the URI to the file being downloaded, and another property that specifies where to write the file to.

        public static DependencyProperty UrlProperty =

                System.Workflow.ComponentModel.DependencyProperty.Register("Url", typeof(Uri), typeof(GetFileActivity));

        [Description("The full URL to the file being requested (ie: ftp://localhost/somefolder/somefile.txt)")]

        [Category("Behavior")]

        [Browsable(true)]

        [DesignerSerializationVisibility(DesignerSerializationVisibility.Visible)]

        public Uri Url

get { return ((Uri)(base.GetValue(GetFileActivity.UrlProperty))); }

            	set{ base.SetValue(GetFileActivity.UrlProperty, value);}

        public static DependencyProperty LocalFilePathProperty = System.Workflow.ComponentModel.DependencyProperty.Register("LocalFilePath", typeof(string), typeof(GetFileActivity));

        [Description("The local file path where the downloaded file will be stored")]

        [Category("Behavior")]

        [Browsable(true)]

        [DesignerSerializationVisibility(DesignerSerializationVisibility.Visible)]

        public string LocalFilePath

            get { return ((string)(base.GetValue(GetFileActivity.LocalFilePathProperty))); }

            set { base.SetValue(GetFileActivity.LocalFilePathProperty, value); }

Now, we need to override the Execute method of the activity and return an ActivityExecutionStatus enumeration member. Here's where the heart of the problem lies: how do we do something asynchronously within an activity? We could use an AutoResetEvent and signal when the work is completed. But WF provides a better way: we can leverage a custom workflow service that will signal our activity when it has completed its work. The activity will simply pass some work to its underlying service and tell the workflow runtime that it is not yet done with its work. The workflow runtime, then, will not complete the workflow until our activity indicates that it has completed (or is aborting for some reason).

Think about how the DelayActivity would work under the covers. The DelayActivity itself doesn't block the current thread, it yields back to the workflow runtime and indicates that it has not yet completed its work. Something else, then, signals the DelayActivity that a specified time period has elapsed, and the DelayActivity completes its work. The DelayActivity leverages a TimerEventSubscription from the WorkflowRuntime, and the runtime signals the DelayActivity when it is to resume its workload.

We will approach this in the same way. We will override the Execute method in our Activity and return the status of Executing to let the workflow runtime know that we are not done with all of our work yet. At the same time, we will also tell our custom service to start doing its work.

        protected override ActivityExecutionStatus Execute(ActivityExecutionContext executionContext)

            WorkflowQueuingService qservice = executionContext.GetService<WorkflowQueuingService>();

            WorkflowQueue q = qservice.GetWorkflowQueue(this.Name);

            q.QueueItemAvailable += this.ContinueAt;

            FtpService s = executionContext.GetService<FtpService>();

            FtpWebRequest request = (FtpWebRequest)FtpWebRequest.Create(this.Url);

            request.Method = WebRequestMethods.Ftp.DownloadFile;

            FtpState state = new FtpState(request,LocalFilePath,this.Name,this.WorkflowInstanceId);

            s.BeginFtp(state);

            return ActivityExecutionStatus.Executing;

Some interesting bits here to pay close attention to. The first is the notion of a WorkflowQueueingService. If you are familiar with MSMQ, then this is very similar in concept. WF under the covers provides a set of queues as a means of communicating between activities, services, and the workflow host. We add a delegate handler called "ContinueAt" that will be called whenever a queue item is available in our workflow queue. We then cruft up some data for an FtpWebRequest, and create a custom type called FtpState. We'll look at this in more detail in a bit, but it is simply a data transfer object to hold a bunch of data related to performing this FTP request. We call our custom FtpService and ask it to start the process, and then we tell the workflow runtime that we are not done with all of our work yet.

We mentioned the ContinueAt method. This is the "bookmark" that we use to continue processing. Our ContinueAt method is called when a workflow queue item is available. We use this to dequeue the item and do any processing. We could do something with the state, such as setting another dependency property that includes exeception data or the FTP status, that task is left to the reader as cleanup.

        void ContinueAt(object sender, QueueEventArgs e)

            ActivityExecutionContext context = sender as ActivityExecutionContext;

            WorkflowQueuingService qservice = context.GetService<WorkflowQueuingService>();

            WorkflowQueue q = qservice.GetWorkflowQueue(this.Name);

            FtpState state = (FtpState)q.Dequeue();

            context.CloseActivity();

The workflow activity framework provides a pipeline of events for processing our activity. Since our activity may be constructed and destructed many times during this lifecycle, we cannot rely on using a constructor for initialization data. An Initialize method is provided for us to perform initialization operations, such as ensuring that our workflow queue exists.

        protected override void Initialize(IServiceProvider provider)

            FtpService s = (FtpService)provider.GetService(typeof(FtpService));

            if (null == s)

                throw new InvalidOperationException("The FtpService has not been added to the workflow");

            //TODO: Could also check for persistence service by looking for

            // WorkflowPersistenceService.  Ommitted for clarity.

            WorkflowQueuingService qservice = (WorkflowQueuingService)provider.GetService(typeof(WorkflowQueuingService));

            if (!qservice.Exists(this.Name))

                WorkflowQueue q = qservice.CreateWorkflowQueue(this.Name, true);

We can also handle the opposite case, tearing down any resources we created during the Initalize operation.

        protected override void Uninitialize(IServiceProvider provider)

            WorkflowQueuingService qservice = (WorkflowQueuingService)provider.GetService(typeof(WorkflowQueuingService));

            if (qservice.Exists(this.Name))

                qservice.DeleteWorkflowQueue(this.Name);

That's our activity in its entirety. Once you break it down, there's not much to it. But how do we actually get work done? The answer lies in the underlying service. There's not much to it, really. We just download a stream and write it to file.

using System;
using System.IO;
using System.Net;
using System.Workflow.Runtime;

using System.Workflow.Runtime.Hosting;

namespace Workflow.Ftp

    public partial class FtpService : WorkflowRuntimeService

        public void BeginFtp(FtpState state)

            state.Request.BeginGetResponse(FtpCallBack, state);

        public void FtpCallBack(IAsyncResult result)

            FtpState state = (FtpState)result.AsyncState;

            FtpWebResponse response = null;

try

                response = (FtpWebResponse)state.Request.EndGetResponse(result);

                using(Stream responseStream = response.GetResponseStream())

                    using(FileStream fileStream = new
				FileStream(state.LocalFilePath, FileMode.OpenOrCreate))

                        CopyStream(responseStream,fileStream);

                response.Close();

                state.StatusDescription = response.StatusDescription;

                //Tell the Activity that we are done with the download

                WorkflowInstance instance = this.Runtime.GetWorkflow(state.WorkflowInstanceId);

                instance.EnqueueItem(state.WorkflowQueueName, state, null, null);

            catch (Exception e)

                state.OperationException = e;

        private void CopyStream(Stream source, Stream target)

            byte[] buffer = new byte[1024];

            int size = 1024;

            int count = 0;

            count = source.Read(buffer, 0, size);

            while (count > 0)

                target.Write(buffer, 0, count);

                count = source.Read(buffer, 0, size);

            target.Flush();

The part to pay attention to is highlighted. Our service obtains a reference to a workflow instance and then enqueues an item into the specified queue. This is how our service will notify our custom activity that the asynchronous work has completed and it can resume where it left off… at its ContinueAt method! A trap that I ran into while trying to figure this out: you might be tempted to create a new WorkflowRuntime and then call its GetWorkflow method. There is an easier way: just inherit your service from WorkflowRuntimeService, and you will have access to the WorkflowRuntime through its inherited member "Runtime". Our FtpService doesn't have to derive from WorkflowRuntimeService, but it sure makes things easier by having access to the WorkflowRuntime through the "Runtime" property.

The final bit of plumbing to show is the FtpState class. Nothing much to tell about here, its just a data transfer object between the activity and the workflow service. I chose to make most of the properties read only (omitting the set accessor) because those properties should not be changed once the FTP process has started.

using System;
using System.Threading;
using System.Net;

namespace Workflow.Ftp

    public class FtpState

        private FtpWebRequest _request;

        private Uri _url;

        private string _localFilePath;

        private Exception operationException = null;

        private string _workflowQueueName;

        private Guid _workflowInstanceId;

        private string status;

        public FtpState(FtpWebRequest request,string localFilePath, string workflowQueueName, Guid workflowInstanceId)

            _request = request;

            _url = request.RequestUri;

            _localFilePath = localFilePath;

            _workflowQueueName = workflowQueueName;

            _workflowInstanceId = workflowInstanceId;

        public FtpWebRequest Request

            get { return _request; }

        public Uri Url

            get { return _url; }

        public string LocalFilePath

            get { return _localFilePath; }

        public Exception OperationException

            get { return operationException; }

            set { operationException = value; }

        public string StatusDescription

            get { return status; }

            set { status = value; }

        public string WorkflowQueueName

            get { return _workflowQueueName; }

        public Guid WorkflowInstanceId

            get { return _workflowInstanceId; }

We will now focus on our Workflow Host. Like any good demo, we will use a Console app. In the Console app, we spin up a workflow runtime and add services to the workflow, same as usual. Only this time, we will add our custom service to the runtime as well, which is highlighted. The host is pretty simple, it's pretty much the default one that Visual Studio will create for you with the addition of our custom service. We also add the SqlWorkflowPersistenceService in there as well, since that is required for our activity.

#region Using directives

using System;

using System.Collections.Generic;

using System.Text;

using System.Threading;

using System.Workflow.Runtime;

using System.Workflow.Runtime.Hosting;

using Workflow.Ftp;

#endregion

namespace WorkflowConsoleApplication1

    class Program

        static AutoResetEvent waitHandle;

        static void Main(string[] args)

            using(WorkflowRuntime workflowRuntime = new
				WorkflowRuntime())

                FtpService s = new FtpService();

                workflowRuntime.AddService(s);

                SqlWorkflowPersistenceService ps = new
				SqlWorkflowPersistenceService("Data Source=(local);Initial Catalog=SqlPersistenceService;Integrated Security=True");

                workflowRuntime.AddService(ps);

                waitHandle = new AutoResetEvent(false);

                workflowRuntime.WorkflowCompleted += new
				EventHandler<WorkflowCompletedEventArgs>(workflowRuntime_WorkflowCompleted);

                workflowRuntime.WorkflowTerminated += delegate(object sender, WorkflowTerminatedEventArgs e)

                    Console.WriteLine(e.Exception.Message);

                    waitHandle.Set();

};

                WorkflowInstance instance = workflowRuntime.CreateWorkflow(typeof(WorkflowConsoleApplication1.Workflow1));

                instance.Start();

                waitHandle.WaitOne();

        static void workflowRuntime_WorkflowCompleted(object sender, WorkflowCompletedEventArgs e)

            waitHandle.Set();

Now that we completed the activity, the service, the asynchronous state object, and our workflow host, let's see how we can use the activity from a workflow. Here's the XOML for the workflow. Pretty uninteresting, other than it completes the story.

<SequentialWorkflowActivity
	x:Class="WorkflowConsoleApplication1.Workflow2"
	x:Name="Workflow2"
	xmlns:ns0="clr-namespace:Workflow.Ftp;Assembly=Workflow.Ftp, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null"
	xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
	xmlns="http://schemas.microsoft.com/winfx/2006/xaml/workflow">

  <ns0:GetFileActivity
	x:Name="Activity1"
	LocalFilePath="c:\temp\testing.txt">

    <ns0:GetFileActivity.Url>

      <ns1:Uri
	xmlns:ns1="clr-namespace:System;Assembly=System, Version=2.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089" />

    </ns0:GetFileActivity.Url>

  </ns0:GetFileActivity>

</SequentialWorkflowActivity>

One caveat: now that I completed this and typed out this ultra-long blog post, I found out that the Workflow designer doesn't like having the System.Uri property, tossing an error "CreateInstance failed for type 'System.Uri'. No parameterless constructor defined for this object." You could rectify this by changing the Url property of the activity back to a string and then parsing it with the System.Uri type to ensure that the scheme is FTP. Even though the designer doesn't like it, the workflow still runs like a champ.

This may look like a bit of code, but there really isn't much to it. Here's a summary of what you can do to process asynchronous operations using a custom workflow activity:

Create a custom activity	Override the Execute method and return ActivityExecutionStatus.Executing. Specify a "ContinueAt" method that will execute when a queue message is received.
Create a custom service	Process the work asynchronously Enqueue a message when the work is completed

This has a lot of applications going forward. For instance, the InvokeWebServiceActivity that ships as part of Windows Workflow Foundation invokes web services synchronously. If the web services are being invoked from ASP.NET pages, then you want to perform that work asynchronously, and the way to achieve that would be to use this pattern. Maybe that's the next article… calling a web service asynchronously. Maybe a few less lines of code, but the pattern will be basically the same.

The sample code is available as a download, attached to this blog post. I welcome feedback, please feel free to leave comments!

Hmm… I wonder… maybe there's a pattern here for generalized activities that support the IAsyncResult pattern?

SYSK 352: Windows Workflow Foundation Runtime Footprint

AppDev: Something You Should Know by Irena Kennedy — 2007-06-26T00:00:00

After seeing presentations that assert that “Workflow runtime is lightweight” and not finding any concrete numbers to support that claim I decided to do some of my own profiling.

Here is what I’ve done, measuring memory utilization after every step (the displayed results are the difference, in bytes, of current memory consumption minus one at the previous check point):

Create a simple Windows app -- one form with a couple of buttons on it.
Get the baseline memory utilization
Instantiate workflow runtime using early binding

System.Workflow.Runtime.WorkflowRuntime wr = new System.Workflow.Runtime.WorkflowRuntime();

Start runtime:

wr.StartRuntime();

Create a basic (nothing except for the initial state) state machine workflow:

wr.CreateWorkflow(typeof(RulesAndWorkflows.Workflow1));

Stop runtime and dispose the object:

wr.StopRuntime();

wr.Dispose();

wr = null;

Here are the early-binding results (in release mode):

Baseline private bytes: 18,194,432
Instantiate runtime: 1,232,896
Start runtime: 1,724,416
Creating empty workflow (loads Activities and ComponentModel assemblies): 2,203,648
Stop runtime, but do not unload assemblies: 0

Note: running same code again results in no additional memory consumption:

Instantiate runtime: 0
Start runtime: 0
Creating empty workflow (loads Activities and ComponentModel assemblies): 0
Stop runtime, but do not unload assemblies: 0

In the second type of test I wanted to profile the memory utilization when loading assemblies “manually”, i.e. using late binding:

Using the same application, get the baseline memory utilization
Instantiate runtime:

AppDomain.CurrentDomain.CreateInstanceAndUnwrap("System.Workflow.Runtime", "System.Workflow.Runtime.WorkflowRuntime");

Load System.Workflow.ComponentModel assembly:

AppDomain.CurrentDomain.Load("System.Workflow.ComponentModel");

Load System.Workflow.Activities assembly:

AppDomain.CurrentDomain.Load("System.Workflow.Activities");

Below are the late-binding results (in release mode, without workflow creation):

Baseline private bytes: 14,028,800
Late-bound runtime instantiation: 4,689,920
Load System.Workflow.ComponentModel assembly: 204,800
Load System.Workflow.Activities assembly: 708,608

Now, armed with data, you can draw your own conclusions about the WF footprint. Of course, keep in mind, that the “real-life” applications will have significantly larger baseline memory utilization than my primitive sample did…

Unit Testing Activity CodeDOM Serializers

ploeh blog — 2007-04-05T00:00:00

Windows Workflow Foundation (WF) defines workflows as object graphs. To save or compile workflow definitions, these object graphs must be serialized, and WF supports serialization to both XAML and code. Similar to other components, WF utilizes the .NET design-time framework for serialization and deserialization of objects. Serializing objects to code is done using CodeDomSerializer-derived classes.

In WF, the base CodeDOM serializer is ActivityCodeDomSerializer, and if you want to unit test your custom CodeDOM serializer for a custom activity, you can use the general approach I have earlier described. As it turns out, unit testing WF Activity CodeDOM serializers is simpler than unit testing components in general, since the base WF classes are already doing some of the work for you. You will still need to implement a designer loader for testing purposes, but you don't need to mess with a fake implementation of ITypeResolutionService.

Let's say that I wish to unit test my custom serialization logic for this activity and its corresponding serializer:

[DesignerSerializer(typeof(MyActivityCodeDomSerializer),

    typeof(CodeDomSerializer))]

public class MyActivity : Activity

    // Implementation goes here

public class MyActivityCodeDomSerializer :

    ActivityCodeDomSerializer

    // Implementation goes here

As you can tell, I've omitted the implementations of both Activity and serializer, since these are not important for this example. Just like the general case, the first thing to do is to create a designer loader that only exists for unit testing purposes. However, instead of deriving from CodeDomDesignerLoader, this class should derive from WorkflowDesignerLoader:

internal class FakeCodeDomDesignerLoader : WorkflowDesignerLoader

    public override string FileName

        get { return "Fake"; }

    public override TextReader GetFileReader(string filePath)

        throw new NotImplementedException();

    public override TextWriter GetFileWriter(string filePath)

        throw new NotImplementedException();

WorkflowDesignerLoader is an abstract class, so it's necessary to override its abstract members, but as you can tell, I haven't done much in that regard. In this case, GetFileReader and GetFileWriter is never called, and although FileName is invoked, the return value is not used for anything that affects the unit test, so I'm just returning a hard-coded string.

When you call BeginLoad on a DesignSurface with this loader (which I'll do in the unit test), PerformLoad will be called, so I have overridden this method to define a default root component:

protected override void PerformLoad(

    IDesignerSerializationManager serializationManager)

    base.PerformLoad(serializationManager);

    CompositeActivity rootActivity = new SequenceActivity();

    this.AddActivityToDesigner(rootActivity);

In this case, the root component will always be a SequenceActivity; this corresponds to opening a new sequential workflow in the Visual Studio designer. You can now add other Activities to the root sequential workflow from the unit test, as I will demonstrate later.

To retrieve the serialized code, the designer loader must be flushed. While flushing, PerformFlush is called, so I override this method to capture the serialized code and save it in a private field:

private CodeStatementCollection codeStatements_;

protected override void PerformFlush(

    IDesignerSerializationManager serializationManager)

    base.PerformFlush(serializationManager);

    ActivityCodeDomSerializer serializer =

        new ActivityCodeDomSerializer();

    this.codeStatements_ =

        (CodeStatementCollection)serializer.Serialize(

        serializationManager, this.LoaderHost.RootComponent);

To retrieve the serialized code from a unit test, I implement a simple method that returns the code statements (recall that calling Flush causes PerformFlush to be called):

internal CodeStatementCollection GetCodeStatements()

    this.Flush();

    return this.codeStatements_;

With FakeCodeDomDesignerLoader in place, it's now fairly straightforward to write a unit test that exercises the custom serializer that's associated with MyActivity:

[TestMethod]

public void MyTestCase()

    FakeCodeDomDesignerLoader loader =

        new FakeCodeDomDesignerLoader();

    DesignSurface surface = new DesignSurface();

    surface.BeginLoad(loader);

    IDesignerHost host =

        (IDesignerHost)surface.GetService(typeof(IDesignerHost));

    MyActivity ma =

        (MyActivity)host.CreateComponent(typeof(MyActivity),

        "ma");

    ((SequenceActivity)host.RootComponent).Activities.Add(ma);

    CodeStatementCollection codeStatements =

        loader.GetCodeStatements();

    // Perform validation of code statements

As with the general approach, the first thing to do is to create an instance of the designer loader. Then I create a new DesignSurface which is going to host the SequenceActivity root component that I can then add an instance of MyActivity to. Calling BeginLoad on the DesignSurface caused PerformLoad to be called on the FakeCodeDomDesignerLoader instance, which then creates the SequenceActivity and adds it as the root component.

The next step is to retrieve an instance of IDesignerHost, which can then be used to create an instance of MyActivity. Creating MyActivity instances this way ensures that any designers, editors and serializers (in this case, MyActivityCodeDomSerializer) is loaded and initialized correctly.

The newly created MyActivity instance is then added to the root component, i.e. the sequential workflow.

Getting the serialized code statements is then as simple as calling GetCodeStatements. This returns a CodeStatementCollection instance, and you can then use the principles outlined in my former article to validate that your custom serializer created the expected code statements.