Event Activities

Previous chapters introduced the concept of event activities. Basically, event activities don’t really do anything except hold the execution of a workflow until something happens. For example, the Delay activity doesn’t do a bit of work aside from waiting a predetermined amount of time before allowing the next activity in the tree to execute. The same goes for the HandleExternalEvent activity, which blocks execution until the configured external event is raised. These event activities implement an interface called IEventActivity.

The IEventActivity Interface

IEventActivity is an interface that, when implemented by a custom activity, can be subscribed to by a listener. The metadata for the IEventActivity interface is shown in the following code:

  namespace System.Workflow.Activities {     public interface IEventActivity     {         IComparable QueueName { get; }         void Subscribe(ActivityExecutionContext parentContext,             IActivityEventListener<QueueEventArgs> parentEventHandler);         void Unsubscribe(ActivityExecutionContext parentContext,             IActivityEventListener<QueueEventArgs> parentEventHandler);     } } namespace System.Workflow.ComponentModel {     public interface IActivityEventListener<T> where T : System.EventArgs     {         void OnEvent(object sender, T e);     } } 

The Subscribe and Unsubscribe events receive the same set of two parameters. The parentContext parameter is of type ActivityExecutionContext and holds a reference to the execution context of the activity that is subscribing an event of interest. The parentEventHandler parameter is of type IActivityEventListener<QueueEventArgs> and is shown the IEventActivity interface’s metadata. This parameter holds a reference to the class that implements the IActivityEventListener <QueueEventArgs> interface and, therefore, has an OnEvent method.

The event handler is raised whenever the corresponding event is raised - more specifically, whenever data arrives in a workflow queue that has already been configured. Queues are discussed in the next section.

Workflow Queues

Workflow queues are at the heart of host-to-workflow communication. To ensure a consistent and robust communication mechanism, queues are used to facilitate the passing of data from the host to a workflow instance. However, data is never passed directly to a workflow instance, although it may appear that way in the code that performs such an operation. Rather, the workflow runtime acts as a mediator to make sure the communication takes place in an orderly fashion.

The WorkflowQueue class encapsulates the queue concept in Windows Workflow Foundation. The following code is a simple example of how to pass data from a workflow runtime host application to a workflow instance:

  WorkflowInstance instance = workflowRuntime.GetWorkflow(instanceId); string myData = "Hello from the outside"; instance.EnqueueItem("MyQueue", myData, null, null); 

The EnqueueItem method of the WorkflowInstance class is responsible for passing data to a workflow. The first parameter is simply the name of the queue to which data is being passed. This queue must have been previously created; otherwise, an InvalidOperationException is thrown. Going back to workflow persistence for a moment, workflow queues created during runtime are persisted when unloading and are re-created during the hydration process.

The data passed in this example happens to be of type string. However, the second parameter is of type object, so anything can be passed as long as it is marked serializable. The final two parameters of the EnqueueItem method are related to work items for transactional execution and are an IPendingWork instance and object instance, respectively.

As shown in the following code, creating a queue and receiving data are almost as easy as sending data to an instance:

  private void CreateQueue(Guid instanceId) {     WorkflowQueuingService queueingService =         workflowRuntime.GetService<WorkflowQueuingService>();     // create the queue     WorkflowQueue queue = queueingService.CreateWorkflowQueue("MyQueue", true);     // we want to be notified when a new piece of data becomes available     queue.QueueItemAvailable += new EventHandler<QueueEventArgs>(itemAvailable) } private void itemAvailable(object sender, QueueEventArgs e) {     WorkflowQueuingService queueingService =         workflowRuntime.GetService<WorkflowQueuingService>();     // get the existing queue     WorkflowQueue queue = queueingService.GetWorkflowQueue(e.QueueName);     string message = (string)queue.Dequeue();     Console.WriteLine("Message was: " + message);     // unsubscribe from this event     queue -= new EventHandler<QueueEventArgs>(itemAvailable);     queueingService.DeleteWorkflowQueue(e.QueueName); } 

First, you can create a new queue by calling the WorkflowQueueingService.CreateWorkflowQueue method. The WorkflowQueueingService can be obtained from the workflow runtime and is added by default. Finally, by subscribing to the QueueItemAvailable event of the WorkflowQueue class, your code is notified whenever an item is added with the EnqueueItem method.

Finally, after the data from the queue is received and the queue is no longer needed, it should unsubscribe from the QueueItemAvailable event. You can then delete the queue by calling the Workflow QueueingService’s DeleteWorkflowQueue method. Before deleting the workflow queue, make sure to retrieve any data by using the Dequeue method of WorkflowQueue.

Developing an Event Activity

To illustrate the development of an event activity, this section creates an example activity that monitors an inbox for new e-mail. This activity could then be used to block workflow execution while waiting for an e-mail and continuing when it is received. The interesting thing about developing activities that implement IEventActivity is that they need to be able to function as children of an EventDriven activity as well as when they are used by themselves. When an event activity is used in an EventDriven activity, which in turn is the child of another activity such as Listen or State, the EventDriven activity parent is responsible for subscribing to the event activity. Conversely, when an event activity is used in a standalone fashion, it needs to be able to subscribe to itself. The good news is that the code needed for both situations is not too complicated and enables you to reuse the methods implemented for the IEventActivity interface.

The Runtime Service

Before digging into the activity code itself, you need to develop a few other support classes for everything to work. For this example, a custom runtime service must be developed to manage e-mail subscriptions. Here is the WaitForEmailService code:

  public class WaitForEmailService : WorkflowRuntimeService {     private List<EmailChecker> watchers = new List<EmailChecker>();     protected override void OnStarted()     {         base.OnStarted();         // start each of the email checkers         foreach (EmailChecker checker in this.watchers)         {             if (!checker.IsStarted) checker.StartChecking();         }     }     protected override void OnStopped()     {         base.OnStopped();         // stop each of the email checkers         foreach (EmailChecker checker in this.watchers)         {             if (checker.IsStarted) checker.PauseChecking();         }     }     public void AddNewWatch(string server, string userName, string password,         int checkFrequencyInSeconds, Guid workflowInstanceId,         IComparable queueName)     {         // here we create a new email checker and subscribe to the         // EmailReceived event         // we also keep track of each subscription in a collection         EmailChecker checker = new EmailChecker(server, userName, password,             checkFrequencyInSeconds, workflowInstanceId, queueName);         checker.EmailReceived +=             new EventHandler<EmailReceivedEventArgs>(checker_EmailReceived);         checker.StartChecking();         this.watchers.Add(checker);     }     private void checker_EmailReceived(object sender, EmailReceivedEventArgs e)     {         // an email has been received! Tell the workflow instance!         EmailChecker checker = sender as EmailChecker;         if (checker != null)         {             WorkflowInstance instance =                 Runtime.GetWorkflow(checker.WorkflowInstanceId);             instance.EnqueueItem(checker.QueueName, e.Message, null, null);             // the email checker is no longer needed             checker.Dispose();         }     } } 

The WaitForEmailService class inherits from WorkflowRuntimeService, which allows it to receive notifications when the workflow runtime is started and stopped. This enables it to respond accordingly - which in this example means starting and pausing e-mail checking.

AddNewWatch is a public method to be called from inside the activity (discussed in more detail later). The parameters include identifying items such as e-mail server information, the workflow instance ID of the workflow that is making the request, and the name of the queue to which messages should be delivered. With this information, a new instance of the EmailChecker class can be created, and the service can subscribe to its EmailReceived event. EmailChecker is a worker class that is responsible for notifying interested parties when new messages arrive. For the purposes of this example, the class does not check a real inbox; instead, it uses a Timer class to simulate an e-mail arriving. The e-mail checker instance is added to a class member collection for later reference.

The event handler for the EmailReceived event is pretty straightforward. It uses the WorkflowRuntime property to obtain a reference to the workflow instance that just received an e-mail and then calls its EnqueueItem method. Because the data passed to this method is accessible to the activity, the Email Message instance is passed as the second parameter. Finally, the EmailChecker is removed from the collection and disposed.

The E-Mail Activity

With the e-mail service developed, the WaitForEmailActivity is ready to be covered. The first part of the code is as follows:

  public class WaitForEmailActivity : Activity, IEventActivity,     IActivityEventListener<QueueEventArgs> {     public IComparable QueueName     {         get { return this.Name + "Queue"; }     }     public void Subscribe(ActivityExecutionContext parentContext,         IActivityEventListener<QueueEventArgs> parentEventHandler)     {         if (parentContext == null)             throw new ArgumentNullException("parentContext");         if (parentEventHandler == null)             throw new ArgumentNullException("parentEventHandler");         // get the email service         WaitForEmailService emailService =             parentContext.GetService<WaitForEmailService>();         if (emailService == null)             throw new InvalidOperationException(                 "The WaitForEmailService is required");         emailService.AddNewWatch(this.Server, this.UserName, this.Password,             this.CheckFrequencyInSeconds, this.WorkflowInstanceId, this.QueueName);         WorkflowQueuingService queuingService =             parentContext.GetService<WorkflowQueuingService>();         WorkflowQueue queue =             queuingService.CreateWorkflowQueue(this.QueueName, false);         queue.RegisterForQueueItemAvailable(parentEventHandler,             base.QualifiedName);         this.isSubscribed = true;     }     public void Unsubscribe(ActivityExecutionContext parentContext,         IActivityEventListener<QueueEventArgs> parentEventHandler)     {         if (parentContext == null)             throw new ArgumentNullException("parentContext");         if (parentEventHandler == null)             throw new ArgumentNullException("parentEventHandler");         WorkflowQueuingService queuingService =             parentContext.GetService<WorkflowQueuingService>();         WorkflowQueue queue = queuingService.GetWorkflowQueue(this.QueueName);         queue.UnregisterForQueueItemAvailable(parentEventHandler);     } } 

The property members represent the required implementation according to the IEventActivity interface. The QueueName property simply provides a unique queue name for a particular workflow instance. Queue names need to be unique only across instances, so you can use the same names in separate instances - the workflow runtime knows which to use.

The Subscribe method is responsible for creating any required queues for communication as well as doing any internal subscription management. The activity uses WaitForEmailService to maintain subscription information for each workflow instance; therefore, the Subscribe method has to ensure that this service has been added to the workflow runtime’s service collection. It is perfectly acceptable to have dependencies on specific runtime services.

Next, a new workflow queue is created, and the event handler passed to the Subscribe method is wired to receive notification whenever a new item is available in the queue. This is done using the WorkflowQueue.RegisterForQueueItemAvailable method. Next, a class field called isSubscribed is set to inform other methods in the class that the subscription is taken care of.

The Unsubscribe method simply obtains a reference to the previously created queue and unsubscribes from the item available event. The queue is not deleted in this step, as you might expect. If the queue is deleted at this point, the activity could not access the data that has been delivered. Rather, the queue is deleted by another method, which is discussed later.

The e-mail activity implements not only IEventActivity, but also the IActivityEventListener <QueueEventArgs> interface. This interface is implemented by a custom activity that needs to be informed when an event activity’s event has occurred. In this case, the e-mail activity is interested when its own event is raised. However, it is important to understand when the implemented OnEvent method is called. When event activities are children of activities such as Listen, the parent is responsible for listening for the event. Therefore, it should be no surprise that these activities also implement IActivityEventListener<QueueEventArgs>. However, when the e-mail activity is used in a standalone manner, it is responsible for listening to itself and doing whatever is necessary to proceed. The following code shows the parts of the e-mail activity that illustrate these concepts:

 public class WaitForEmailActivity : Activity, IEventActivity,     IActivityEventListener<QueueEventArgs> {     void IActivityEventListener<QueueEventArgs>.OnEvent(object sender,         QueueEventArgs e)     {         ActivityExecutionContext context = sender as ActivityExecutionContext;         if (context != null &&             this.ExecutionStatus == ActivityExecutionStatus.Executing)         {             if (this.CheckQueue(context) == ActivityExecutionStatus.Closed)             {                 context.CloseActivity();             }         }     }      private ActivityExecutionStatus CheckQueue(ActivityExecutionContext context)     {         if (this.isSubscribed)         {             WorkflowQueuingService queuingService =                 context.GetService<WorkflowQueuingService>();             WorkflowQueue queue = queuingService.GetWorkflowQueue(this.QueueName);              if (queue.Count > 0)             {                 object data = queue.Dequeue();                 if (!(data is EmailMessage))                     throw new InvalidOperationException(                         "This activity can only handle EmailMessage objects");                  base.SetValue(MessageProperty, data);                  // we're done with the queue, go ahead and delete it now                 queuingService.DeleteWorkflowQueue(this.QueueName);                  // we're done, so return a status of closed                 return ActivityExecutionStatus.Closed;             }         }          // not done yet, return an Executing status         return ActivityExecutionStatus.Executing;     } }

The OnEvent method performs a couple of checks up front to make sure everything is how it should be. It checks to make sure that the sender parameter is an ActivityExecutionContext instance, that the activity’s execution status is Executing, and that the CheckQueue method returns a status of Closed. Because of the way the activity executes when not in a subscribing parent, all of these conditions should be true every time this method is called. The CloseActivity method of the context object is called to signal when the activity is done.

The CheckQueue method is next and is where most of the interesting work is performed in this activity. First, if the isSubscribed field is not set, no work is done, and a status of Executing is returned. However, the subscription has already been established, so an instance of the WorkflowQueueingService is acquired, which then allows access to the communication queue. Next, the queue is checked to make sure it has some data available - and the way the activity has been coded, this check should always evaluate to true. If the data dequeued is an EmailMessage instance, the activity’s dependency property, MessageProperty, is set using the SetValue method. Finally, the message queue is deleted, and a status of Closed is returned.

The WaitForEmailActivity’s Execute method is as follows:

 public class WaitForEmailActivity : Activity, IEventActivity,     IActivityEventListener<QueueEventArgs> {     protected override ActivityExecutionStatus Execute(         ActivityExecutionContext context)     {         if (context == null)             throw new ArgumentNullException("context");          // subscribe will only be called here is we are not         // in an activity like ListenActivity or StateActivity         // i.e. if it is being used as a stand along activity         if(!this.isSubscribed)             this.Subscribe(context, this);          return CheckQueue(context);     } }

The activity’s Execute override is surprisingly simple. First, a check against the isSubscribed field is performed. Remember, you need to manually subscribe to the workflow queue only when the activity is not a child to a subscribing parent, such as the Listen activity. The CheckQueue method is then called to get the return status. If this is a standalone activity, Executing is returned; otherwise, a status of Closed is returned, and the activity is considered completed.

Testing the Activity’s Execution

Now that WaitForEmailActivity is completed, it is interesting to see how it executes when used in different workflow scenarios. The execution path and order of method calls differ considerably, depending on whether the activity is in a subscribing parent. Figure 8-2 shows a workflow that has the e-mail activity as a child to a Listen activity. Because of this relationship, the Listen activity is responsible for listening for the event raised when a new item is added to the workflow queue.

Figure 8-2

If you place strategic breakpoints in the code developed for this example, you can see an execution path like the following:

1. Subscribe is called (by the Listen activity), and a workflow queue is created and subscribed to.

2. The workflow goes idle while waiting for an e-mail.

3. An e-mail arrives, thereby notifying the e-mail runtime service, which adds an item to the queue.

4. Unsubscribe is called (by the Listen activity), and the subscription to the workflow queue is undone.

5. Execute is called, and the e-mail message is retrieved from the queue, which is then deleted.

6. A status of Closed is returned, and the workflow proceeds.

Figure 8-3 shows a workflow that uses the e-mail activity in a standalone fashion. As you know by now, this means that the activity is responsible for listening to itself when an item arrives in the queue rather than some other parent.

Figure 8-3

Again, if you use breakpoints and debug the workflow, you can see the following execution path:

1. Execute is called, which in turn calls the Subscribe method. A status of Executing is returned so that the activity does not close.

2. The workflow goes idle while waiting for an e-mail.

3. An e-mail arrives, thereby notifying the e-mail runtime service, which adds an item to the queue.

4. The activity’s OnEvent method is called, which causes the e-mail to be retrieved from the queue. The queue is then deleted.

5. The ActivityExecutionContext.CloseActivity method is called, allowing the workflow to continue.

Other Considerations

Although WaitForEmailActivity is a fairly comprehensive example of how to implement an IEventActivity activity, a few other things could have been done to make it a little better and more robust. First, consider the workflow shown in Figure 8-2 earlier in this chapter. If the Delay activity in the second branch times out before the e-mail arrives, the workflow queue used in the e-mail activity is not deleted. Therefore, it would probably be a good idea to implement code in the activity’s cancellation handler to clean up any resources.

In addition, a more interesting activity might look for very specific e-mails related only to the workflow instance in question. For example, perhaps all workflow instances are checking the same inbox. With the current implementation, only the instance that checks for e-mail first gets the notification. However, code could be implemented to correlate e-mails to a workflow instance. Perhaps each e-mail’s subject is required to contain some GUID that ties back to a workflow instance. Regular expressions could be used to parse the text and raise the EmailReceived event only when appropriate.