diff --git a/includes/rules.core.inc b/includes/rules.core.inc
index 3acbbbd..0e7e625 100644
--- a/includes/rules.core.inc
+++ b/includes/rules.core.inc
@@ -132,12 +132,35 @@ class RulesEntityController extends EntityAPIControllerExportable {
   public function save($rules_config, DatabaseTransaction $transaction = NULL) {
     $transaction = isset($transaction) ? $transaction : db_transaction();
 
+    // Load the stored entity, if any.
+    if (!isset($rules_config->original) && $rules_config->{$this->idKey}) {
+      $rules_config->original = entity_load_unchanged($this->entityType, $rules_config->{$this->idKey});
+    }
+    $original = isset($rules_config->original) ? $rules_config->original : NULL;
+
     $return = parent::save($rules_config, $transaction);
     $this->storeTags($rules_config);
     if ($rules_config instanceof RulesTriggerableInterface) {
       $this->storeEvents($rules_config);
     }
     $this->storeDependencies($rules_config);
+
+    // See if there are any events that have been removed.
+    if ($original && $rules_config->plugin == 'reaction rule') {
+      foreach (array_diff($original->events(), $rules_config->events()) as $event_name) {
+        // Check if the event handler implements the event dispatcher interface.
+        $handler = rules_get_event_handler($event_name, $rules_config->getEventSettings($event_name));
+        if (!$handler instanceof RulesEventDispatcherInterface) {
+          continue;
+        }
+
+        // Only stop an event dispatcher if there are no rules for it left.
+        if (!rules_config_load_multiple(FALSE, array('event' => $event_name, 'plugin' => 'reaction rule', 'active' => TRUE)) && $handler->isWatching()) {
+          $handler->stopWatching();
+        }
+      }
+    }
+
     return $return;
   }
 
@@ -239,7 +262,36 @@ class RulesEntityController extends EntityAPIControllerExportable {
           ->execute();
       }
     }
-    return parent::delete($ids, $transaction);
+    $return = parent::delete($ids, $transaction);
+
+    // Stop event dispatchers when deleting the last rule of an event set.
+    $processed = array();
+    foreach ($configs as $config) {
+      if (!isset($config->plugin) || $config->plugin != 'reaction rule') {
+        continue;
+      }
+
+      foreach ($config->events() as $event_name) {
+        // Only process each event once.
+        if (in_array($event_name, $processed)) {
+          continue;
+        }
+        $processed[$event_name] = TRUE;
+
+        // Check if the event handler implements the event dispatcher interface.
+        $handler = rules_get_event_handler($event_name, $config->getEventSettings($event_name));
+        if (!$handler instanceof RulesEventDispatcherInterface) {
+          continue;
+        }
+
+        // Only stop an event dispatcher if there are no rules for it left.
+        if (!rules_config_load_multiple(FALSE, array('event' => $event_name, 'plugin' => 'reaction rule', 'active' => TRUE)) && $handler->isWatching()) {
+          $handler->stopWatching();
+        }
+      }
+    }
+
+    return $return;
   }
 }
 
diff --git a/includes/rules.event.inc b/includes/rules.event.inc
index 893018d..257acce 100755
--- a/includes/rules.event.inc
+++ b/includes/rules.event.inc
@@ -125,6 +125,46 @@ interface RulesEventHandlerInterface {
    *   variable info array as value.
    */
   public function availableVariables();
+
+  /**
+   * Returns the name of the event the event handler belongs to.
+   *
+   * @return string
+   *   The name of the event the event handler belongs to.
+   */
+  public function getEventName();
+
+  /**
+   * Returns the info array of the event the event handler belongs to.
+   *
+   * @return string
+   *   The info array of the event the event handler belongs to.
+   */
+  public function getEventInfo();
+}
+
+/**
+ * Interface for event dispatchers.
+ */
+interface RulesEventDispatcherInterface extends RulesEventHandlerInterface {
+
+  /**
+   * Starts the event watcher.
+   */
+  public function startWatching();
+
+  /**
+   * Stops the event watcher.
+   */
+  public function stopWatching();
+
+  /**
+   * Returns whether the event dispatcher is currently active.
+   *
+   * @return bool
+   *   TRUE if the event dispatcher is currently active, FALSE otherwise.
+   */
+  public function isWatching();
 }
 
 /**
@@ -199,6 +239,20 @@ abstract class RulesEventHandlerBase implements RulesEventHandlerInterface {
   public function availableVariables() {
     return isset($this->eventInfo['variables']) ? $this->eventInfo['variables'] : array();
   }
+
+  /**
+   * Implements RulesEventHandlerInterface::getEventName()
+   */
+  public function getEventName() {
+    return $this->eventName;
+  }
+
+  /**
+   * Implements RulesEventHandlerInterface::getEventInfo()
+   */
+  public function getEventInfo() {
+    return $this->eventInfo;
+  }
 }
 
 /**
diff --git a/includes/rules.plugins.inc b/includes/rules.plugins.inc
index 29778ec..85b7296 100644
--- a/includes/rules.plugins.inc
+++ b/includes/rules.plugins.inc
@@ -773,6 +773,12 @@ class RulesEventSet extends RulesRuleSet {
         // Create an event set if not yet done.
         if (!isset($sets[$event_name])) {
           $handler = rules_get_event_handler($event_name, $rule->getEventSettings($event_name));
+
+          // Start the event dispatcher for this event, if any.
+          if ($handler instanceof RulesEventDispatcherInterface && !$handler->isWatching()) {
+            $handler->startWatching();
+          }
+
           // Update the event info with the variables available based on the
           // event settings.
           $event_info = $events[$event_base_name];
diff --git a/rules.info b/rules.info
index 9e3b2d0..d7ab7fe 100644
--- a/rules.info
+++ b/rules.info
@@ -10,6 +10,7 @@ files[] = includes/rules.event.inc
 files[] = includes/rules.processor.inc
 files[] = includes/rules.plugins.inc
 files[] = includes/rules.state.inc
+files[] = includes/rules.dispatcher.inc
 files[] = modules/php.eval.inc
 files[] = modules/rules_core.eval.inc
 files[] = modules/system.eval.inc
diff --git a/rules_scheduler/includes/rules_scheduler.handler.inc b/rules_scheduler/includes/rules_scheduler.handler.inc
new file mode 100644
index 0000000..83cb66d
--- /dev/null
+++ b/rules_scheduler/includes/rules_scheduler.handler.inc
@@ -0,0 +1,89 @@
+<?php
+
+/**
+ * Default scheduled task handler.
+ */
+class RulesSchedulerDefaultTaskHandler implements RulesSchedulerTaskHandlerInterface {
+
+  /**
+   * Constructs a repetitive task handler object.
+   */
+  public function __construct(array $task) {
+    $this->task = $task;
+  }
+
+  /**
+   * Implements RulesSchedulerTaskHandlerInterface::runTask().
+   */
+  public function runTask() {
+    if ($component = rules_get_cache('comp_' . $this->task['config'])) {
+      $replacements = array('%label' => $component->label(), '%plugin' => $component->plugin());
+      $replacements['%identifier'] = $this->task['identifier'] ? $this->task['identifier'] : t('without identifier');
+      rules_log('Scheduled evaluation of %plugin %label, task %identifier.', $replacements, RulesLog::INFO, $component, TRUE);
+      $state = unserialize($this->task['data']);
+      $state->restoreBlocks();
+      // Block the config to prevent any future recursion.
+      $state->block($component);
+      // Finally evaluate the component with the given state.
+      $component->evaluate($state);
+      $state->unblock($component);
+      rules_log('Finished evaluation of %plugin %label, task %identifier.', $replacements, RulesLog::INFO, $component, FALSE);
+      $state->cleanUp();
+    }
+  }
+
+  /**
+   * Implements RulesSchedulerTaskHandlerInterface::afterTaskQueued().
+   */
+  public function afterTaskQueued() {
+    // Delete the task from the task list.
+    db_delete('rules_scheduler')
+      ->condition('tid', $this->task['tid'])
+      ->execute();
+  }
+
+  /**
+   * Implements RulesSchedulerTaskHandlerInterface::getTask().
+   */
+  public function getTask() {
+    return $this->task;
+  }
+
+}
+
+/**
+ * Interface for scheduled task handlers.
+ *
+ * Task handlers control the behavior of a task when it's queued or executed.
+ * Unless specified otherwise, the RulesSchedulerDefaultTaskHandler task handler
+ * is used.
+ *
+ * @see rules_scheduler_run_task()
+ * @see rules_scheduler_cron()
+ * @see RulesSchedulerDefaultTaskHandler
+ */
+interface RulesSchedulerTaskHandlerInterface {
+
+  /**
+   * Processes a queue item.
+   *
+   * @see rules_scheduler_run_task()
+   */
+  public function runTask();
+
+  /**
+   * Processes a task after it has been queued.
+   *
+   * @see rules_scheduler_cron()
+   */
+  public function afterTaskQueued();
+
+  /**
+   * Returns the task associated with the task handler.
+   *
+   * @return array
+   *   The task (queue item) array.
+   */
+  public function getTask();
+
+}
diff --git a/rules_scheduler/rules_scheduler.info b/rules_scheduler/rules_scheduler.info
index e1d8a07..68db15c 100644
--- a/rules_scheduler/rules_scheduler.info
+++ b/rules_scheduler/rules_scheduler.info
@@ -8,6 +8,7 @@ files[] = rules_scheduler.module
 files[] = rules_scheduler.install
 files[] = rules_scheduler.rules.inc
 files[] = rules_scheduler.test
+files[] = includes/rules_scheduler.handler.inc
 files[] = includes/rules_scheduler.views_default.inc
 files[] = includes/rules_scheduler.views.inc
 files[] = includes/rules_scheduler_views_filter.inc
diff --git a/rules_scheduler/rules_scheduler.install b/rules_scheduler/rules_scheduler.install
index 7420040..982dcf7 100644
--- a/rules_scheduler/rules_scheduler.install
+++ b/rules_scheduler/rules_scheduler.install
@@ -30,11 +30,11 @@ function rules_scheduler_schema() {
         'type' => 'int',
         'not null' => TRUE,
       ),
-      'state' => array(
+      'data' => array(
         'type' => 'text',
         'not null' => FALSE,
         'serialize' => TRUE,
-        'description' => 'The whole, serialized evaluation state.',
+        'description' => 'The whole, serialized evaluation data.',
       ),
       'identifier' => array(
         'type' => 'varchar',
@@ -43,6 +43,12 @@ function rules_scheduler_schema() {
         'not null' => FALSE,
         'description' => 'The user defined string identifying this task.',
       ),
+      'handler' => array(
+        'type' => 'varchar',
+        'length' => '255',
+        'not null' => FALSE,
+        'description' => 'The fully qualified class name of a the queue item handler.',
+      ),
     ),
     'primary key' => array('tid'),
     'indexes' => array(
@@ -84,11 +90,11 @@ function rules_scheduler_update_7200() {
         'type' => 'int',
         'not null' => TRUE,
       ),
-      'state' => array(
+      'data' => array(
         'type' => 'text',
         'not null' => FALSE,
         'serialize' => TRUE,
-        'description' => 'The whole, serialized evaluation state.',
+        'description' => 'The whole, serialized evaluation data.',
       ),
       'identifier' => array(
         'type' => 'varchar',
@@ -123,6 +129,30 @@ function rules_scheduler_update_7202() {
 }
 
 /**
+ * Add a database column for specifying a queue item handler.
+ */
+function rules_scheduler_update_7203() {
+  db_add_field('rules_scheduler', 'handler', array(
+    'type' => 'varchar',
+    'length' => '255',
+    'not null' => FALSE,
+    'description' => 'The fully qualified class name of a the queue item handler.',
+  ));
+}
+
+/**
+ * Rename rules_scheduler.state into rules_scheduler.data.
+ */
+function rules_scheduler_update_7204() {
+  db_change_field('rules_scheduler', 'state', 'data', array(
+    'type' => 'text',
+    'not null' => FALSE,
+    'serialize' => TRUE,
+    'description' => 'The whole, serialized evaluation data.',
+  ));
+}
+
+/**
  * Rules upgrade callback for mapping the action name.
  */
 function rules_scheduler_action_upgrade_map_name($element) {
diff --git a/rules_scheduler/rules_scheduler.module b/rules_scheduler/rules_scheduler.module
index 8e6b0dd..2aa5e47 100644
--- a/rules_scheduler/rules_scheduler.module
+++ b/rules_scheduler/rules_scheduler.module
@@ -15,6 +15,7 @@ function rules_scheduler_cron() {
   $result = db_select('rules_scheduler', 'r', array('fetch' => PDO::FETCH_ASSOC))
     ->fields('r')
     ->condition('date', time(), '<=')
+    ->orderBy('date')
     ->range(0, 1000)
     ->execute();
 
@@ -22,10 +23,8 @@ function rules_scheduler_cron() {
   foreach ($result as $task) {
     // Add the task to the queue and remove the entry afterwards.
     if ($queue->createItem($task)) {
-      db_delete('rules_scheduler')
-        ->condition('tid', $task['tid'])
-        ->execute();
       $task_created = TRUE;
+      rules_scheduler_task_handler($task)->afterTaskQueued();
     }
   }
 
@@ -52,22 +51,25 @@ function rules_scheduler_cron_queue_info() {
 
 /**
  * Queue worker callback for running a single task.
+ *
+ * @param array $task
+ *   The task to process.
  */
 function rules_scheduler_run_task(array $task) {
-  if ($component = rules_get_cache('comp_' . $task['config'])) {
-    $replacements = array('%label' => $component->label(), '%plugin' => $component->plugin());
-    $replacements['%identifier'] = $task['identifier'] ? $task['identifier'] : t('without identifier');
-    rules_log('Scheduled evaluation of %plugin %label, task %identifier.', $replacements, RulesLog::INFO, $component, TRUE);
-    $state = unserialize($task['state']);
-    $state->restoreBlocks();
-    // Block the config to prevent any future recursion.
-    $state->block($component);
-    // Finally evaluate the component with the given state.
-    $component->evaluate($state);
-    $state->unblock($component);
-    rules_log('Finished evaluation of %plugin %label, task %identifier.', $replacements, RulesLog::INFO, $component, FALSE);
-    $state->cleanUp();
-  }
+  rules_scheduler_task_handler($task)->runTask();
+}
+
+/**
+ * Returns the task handler for a given task.
+ *
+ * @param array $task
+ *   A task (queue item) array.
+ * @return RulesSchedulerTaskHandlerInterface
+ *   The task handler.
+ */
+function rules_scheduler_task_handler(array $task) {
+  $class = !empty($task['handler']) ? $task['handler'] : 'RulesSchedulerDefaultTaskHandler';
+  return new $class($task);
 }
 
 /**
@@ -138,7 +140,7 @@ function rules_scheduler_task_delete($tid) {
  *   An array representing the task with the following keys:
  *   - config: The machine readable name of the to be scheduled component.
  *   - date: Timestamp when the component should be executed.
- *   - state: An rules evaluation state to use for scheduling.
+ *   - data: Rules evaluation data to use for scheduling.
  *   - identifier: User provided string to identify the task per scheduled
  *   configuration.
  */
diff --git a/rules_scheduler/rules_scheduler.rules.inc b/rules_scheduler/rules_scheduler.rules.inc
index dfcda86..473a810 100644
--- a/rules_scheduler/rules_scheduler.rules.inc
+++ b/rules_scheduler/rules_scheduler.rules.inc
@@ -89,7 +89,7 @@ function rules_scheduler_action_schedule($args, $element) {
     rules_scheduler_schedule_task(array(
       'date' => $args['date'],
       'config' => $args['component'],
-      'state' => $new_state,
+      'data' => $new_state,
       'identifier' => $args['identifier'],
     ));
   }