Issue #3215938 by paul121: Data stream email notifications

modules/core/data_stream/modules/notification/config/schema/data_stream_notification.condition.schema.yml

+# Schema for the data stream notification condition plugins.
+
+data_stream_notification.condition.numeric:
+  type: data_stream_notification_condition
+  label: 'Numeric condition'
+  mapping:
+    condition:
+      type: string
+      label: 'Condition'
+    threshold:
+      type: float
+      label: 'Threshold'

modules/core/data_stream/modules/notification/config/schema/data_stream_notification.data_types.schema.yml

+# Basic data types for data stream notification.
+
+data_stream_notification_condition:
+  type: mapping
+  label: 'Condition configuration'
+  mapping:
+    type:
+      type: string
+      label: 'Condition plugin'
+    negate:
+      type: boolean
+      label: 'Negate'
+
+data_stream_notification_delivery:
+  type: mapping
+  label: 'Delivery configuration'
+  mapping:
+    type:
+      type: string
+      label: 'Delivery plugin'

modules/core/data_stream/modules/notification/config/schema/data_stream_notification.delivery.schema.yml

+# Schema for the data stream notification delivery plugins.
+
+data_stream_notification.delivery.email:
+  type: data_stream_notification_delivery
+  mapping:
+    email:
+      type: sequence
+      label: 'Emails'
+      sequence:
+        label: 'Email'
+        type: email

modules/core/data_stream/modules/notification/config/schema/data_stream_notification.schema.yml

+# Schema for the configuration files of the Data stream notifications module.
+
+data_stream_notification.data_stream_notification.*:
+  type: config_entity
+  label: 'Data stream notification'
+  mapping:
+    id:
+      type: string
+      label: 'ID'
+    label:
+      type: label
+      label: 'Label'
+    data_stream:
+      type: integer
+      nullable: true
+      label: 'Data stream'
+    activation_threshold:
+      type: integer
+      label: 'Activation threshold'
+    deactivation_threshold:
+      type: integer
+      label: 'Deactivation threshold'
+    condition_operator:
+      type: string
+      label: 'Condition operator'
+    condition:
+      type: sequence
+      label: 'Conditions'
+      sequence:
+        type: data_stream_notification.condition.[type]
+    delivery_interval:
+      type: integer
+      label: 'Delivery interval'
+    delivery:
+      type: sequence
+      label: 'Delivery'
+      sequence:
+        type: data_stream_notification.delivery.[type]

modules/core/data_stream/modules/notification/data_stream_notification.info.yml

+name: farmOS Data Stream Notification
+description: Provides notifications for data streams.
+type: module
+package: farmOS
+core_version_requirement: ^9
+dependencies:
+  - farm:data_stream

modules/core/data_stream/modules/notification/data_stream_notification.links.action.yml

+data_stream_notification.add:
+  route_name: entity.data_stream_notification.add_form
+  title: 'Add notification'
+  appears_on:
+    - entity.data_stream_notification.collection

modules/core/data_stream/modules/notification/data_stream_notification.module

+<?php
+
+/**
+ * @file
+ * Contains data_stream_notification.module.
+ */
+
+/**
+ * Implements hook_mail().
+ */
+function data_stream_notification_mail($key, &$message, $params) {
+
+  // Bail if not a notification email.
+  if ($key !== 'notification_email') {
+    return;
+  }
+
+  /** @var \Drupal\data_stream_notification\Entity\DataStreamNotification $data_stream_notification */
+  $data_stream_notification = $params['data_stream_notification'];
+
+  /** @var \Drupal\data_stream\Entity\DataStreamInterface $data_stream */
+  $data_stream = $params['data_stream'];
+  $url = $data_stream->toUrl()->setAbsolute()->toString();
+
+  // Build the email subject.
+  $message['subject'] = t(
+    '@notification notification for data stream: @data_stream',
+    [
+      '@notification' => $data_stream_notification->label(),
+      '@data_stream' => $data_stream->label(),
+    ]
+  );
+
+  // Build the email body.
+  $message['body'][] = t(
+    'Data stream: <a href="@link">@label</a> Actual value: @value',
+    [
+      '@link' => $url,
+      '@label' => $data_stream->label(),
+      '@value' => $params['value'],
+    ],
+  );
+
+  // Build list of condition summaries.
+  $conditions = '<ul>';
+  $conditions .= '<li>' . implode('</li><li>', $params['condition_summaries']) . '</li>';
+  $conditions .= '</ul>';
+  $message['body'][] = $conditions;
+}

modules/core/data_stream/modules/notification/data_stream_notification.routing.yml

+entity.data_stream_notification.enable:
+  path: '/data-stream-notifications/{data_stream_notification}/enable'
+  defaults:
+    _controller: '\Drupal\data_stream_notification\Controller\NotificationUIController::ajaxOperation'
+    op: enable
+  requirements:
+    _entity_access: data_stream_notification.edit
+    _csrf_token: 'TRUE'
+
+entity.data_stream_notification.disable:
+  path: '/data-stream-notifications/{data_stream_notification}/disable'
+  defaults:
+    _controller: '\Drupal\data_stream_notification\Controller\NotificationUIController::ajaxOperation'
+    op: disable
+  requirements:
+    _entity_access: data_stream_notification.edit
+    _csrf_token: 'TRUE'

modules/core/data_stream/modules/notification/data_stream_notification.services.yml

+services:
+  plugin.manager.data_stream_notification_condition:
+    class: Drupal\data_stream_notification\NotificationConditionManager
+    parent: default_plugin_manager
+  plugin.manager.data_stream_notification_delivery:
+    class: Drupal\data_stream_notification\NotificationDeliveryManager
+    parent: default_plugin_manager
+  data_stream_notification.data_stream_event_subscriber:
+    class: Drupal\data_stream_notification\EventSubscriber\DataStreamEventSubscriber
+    arguments:
+      [ '@entity_type.manager' ]
+    tags:
+      - { name: 'event_subscriber' }

modules/core/data_stream/modules/notification/src/Annotation/NotificationCondition.php

+<?php
+
+namespace Drupal\data_stream_notification\Annotation;
+
+use Drupal\Component\Annotation\Plugin;
+
+/**
+ * Defines a notification condition annotation object.
+ *
+ * @Annotation
+ */
+class NotificationCondition extends Plugin {
+
+  /**
+   * The condition ID.
+   *
+   * @var string
+   */
+  public $id;
+
+  /**
+   * The condition label.
+   *
+   * @var \Drupal\Core\Annotation\Translation
+   *
+   * @ingroup plugin_translatable
+   */
+  public $label;
+
+}

modules/core/data_stream/modules/notification/src/Annotation/NotificationDelivery.php

+<?php
+
+namespace Drupal\data_stream_notification\Annotation;
+
+use Drupal\Component\Annotation\Plugin;
+
+/**
+ * Defines a notification delivery annotation object.
+ *
+ * @Annotation
+ */
+class NotificationDelivery extends Plugin {
+
+  /**
+   * The delivery ID.
+   *
+   * @var string
+   */
+  public $id;
+
+  /**
+   * The delivery label.
+   *
+   * @var \Drupal\Core\Annotation\Translation
+   *
+   * @ingroup plugin_translatable
+   */
+  public $label;
+
+}

modules/core/data_stream/modules/notification/src/Controller/NotificationUIController.php

+<?php
+
+namespace Drupal\data_stream_notification\Controller;
+
+use Drupal\Core\Ajax\AjaxResponse;
+use Drupal\Core\Ajax\ReplaceCommand;
+use Drupal\Core\Controller\ControllerBase;
+use Drupal\data_stream_notification\Entity\DataStreamNotificationInterface;
+use Symfony\Component\HttpFoundation\Request;
+
+/**
+ * Returns responses for notification UI routes.
+ *
+ * @see \Drupal\views_ui\Controller\ViewsUIController::ajaxOperation()
+ */
+class NotificationUIController extends ControllerBase {
+
+  /**
+   * Calls a method on a data stream notification and reloads the listing page.
+   *
+   * @param \Drupal\data_stream_notification\Entity\DataStreamNotificationInterface $data_stream_notification
+   *   The data stream notification entity.
+   * @param string $op
+   *   The operation to perform, e.g., 'enable' or 'disable'.
+   * @param \Symfony\Component\HttpFoundation\Request $request
+   *   The current request.
+   *
+   * @return \Drupal\Core\Ajax\AjaxResponse|\Symfony\Component\HttpFoundation\RedirectResponse
+   *   Either returns a rebuilt listing page as an AJAX response, or redirects
+   *   back to the listing page.
+   */
+  public function ajaxOperation(DataStreamNotificationInterface $data_stream_notification, string $op, Request $request) {
+    // Perform the operation.
+    $data_stream_notification->$op()->save();
+
+    // Reset the notification state.
+    $data_stream_notification->resetState();
+
+    // If the request is via AJAX, return the rendered list as JSON.
+    if ($request->request->get('js')) {
+      $list = $this->entityTypeManager()
+        ->getListBuilder('data_stream_notification')
+        ->render();
+      $response = new AjaxResponse();
+      $response->addCommand(new ReplaceCommand('#data-stream-notification-entity-list', $list));
+      return $response;
+    }
+
+    // Otherwise, redirect back to the listing page.
+    return $this->redirect('entity.data_stream_notification.collection');
+  }
+
+}

modules/core/data_stream/modules/notification/src/DataStreamNotificationListBuilder.php

+<?php
+
+namespace Drupal\data_stream_notification;
+
+use Drupal\Core\Config\Entity\ConfigEntityListBuilder;
+use Drupal\Core\Entity\EntityInterface;
+
+/**
+ * Provides a data stream notification list builder.
+ */
+class DataStreamNotificationListBuilder extends ConfigEntityListBuilder {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function load() {
+    $entities = [
+      'enabled' => [],
+      'disabled' => [],
+    ];
+    foreach (parent::load() as $entity) {
+      if ($entity->status()) {
+        $entities['enabled'][] = $entity;
+      }
+      else {
+        $entities['disabled'][] = $entity;
+      }
+    }
+    return $entities;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function buildRow(EntityInterface $notification) {
+    $row = parent::buildRow($notification);
+    return [
+      'data' => [
+        'label' => [
+          'data' => [
+            '#plain_text' => $notification->label(),
+          ],
+        ],
+        'machine_name' => [
+          'data' => [
+            '#plain_text' => $notification->id(),
+          ],
+        ],
+        'active' => [
+          'data' => [
+            '#plain_text' => $notification->isActive() ? $this->t('True') : $this->t('False'),
+          ],
+        ],
+        'operations' => $row['operations'],
+      ],
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function buildHeader() {
+    return [
+      'label' => [
+        'data' => $this->t('Label'),
+      ],
+      'machine_name' => [
+        'data' => $this->t('Machine name'),
+      ],
+      'active' => [
+        'data' => $this->t('Active'),
+      ],
+      'operations' => [
+        'data' => $this->t('Operations'),
+      ],
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getDefaultOperations(EntityInterface $entity) {
+    $operations = parent::getDefaultOperations($entity);
+
+    // Add AJAX functionality to enable/disable operations.
+    foreach (['enable', 'disable'] as $op) {
+      if (isset($operations[$op])) {
+        $operations[$op]['url'] = $entity->toUrl($op);
+        // Enable and disable operations should use AJAX.
+        $operations[$op]['attributes']['class'][] = 'use-ajax';
+      }
+    }
+
+    // We assign data-drupal-selector to every link, so it focuses on the edit
+    // link after the ajax response. By default ajax.js would focus on the same
+    // button again, but the enable/disable buttons will be hidden.
+    // @see ViewsListBuilder::getDefaultOperations()
+    foreach ($operations as &$operation) {
+      $operation['attributes']['data-drupal-selector'] = 'data-stream-notification-listing-' . $entity->id();
+    }
+
+    return $operations;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function render() {
+
+    // Render a wrapper container that can be replaced.
+    $list['#type'] = 'container';
+    $list['#attributes']['id'] = 'data-stream-notification-entity-list';
+
+    // Add markup for the enabled table.
+    $list['enabled']['heading']['#markup'] = '<h2>' . $this->t('Enabled', [], ['context' => 'Plural']) . '</h2>';
+    $list['enabled']['table']['#empty'] = $this->t('There are no enabled notifications.');
+
+    // Add markup for the disabled table.
+    $list['disabled']['heading']['#markup'] = '<h2>' . $this->t('Disabled', [], ['context' => 'Plural']) . '</h2>';
+    $list['disabled']['table']['#empty'] = $this->t('There are no disabled notifications.');
+
+    // Build separate tables for enabled and disabled.
+    $entities = $this->load();
+    foreach (['enabled', 'disabled'] as $status) {
+      $list[$status]['table'] = [
+        '#type' => 'table',
+        '#header' => $this->buildHeader(),
+      ];
+
+      // Build a row for each entity.
+      foreach ($entities[$status] as $entity) {
+        if ($row = $this->buildRow($entity)) {
+          $list[$status]['table']['#rows'][$entity->id()] = $row;
+        }
+      }
+    }
+
+    return $list;
+  }
+
+}

modules/core/data_stream/modules/notification/src/DataStreamNotificationPluginCollection.php

+<?php
+
+namespace Drupal\data_stream_notification;
+
+use Drupal\Core\Plugin\DefaultLazyPluginCollection;
+
+/**
+ * A plugin collection for notification condition and delivery plugins.
+ *
+ * Overrides the DefaultLazyPluginCollection to use "type" as the plugin key.
+ */
+class DataStreamNotificationPluginCollection extends DefaultLazyPluginCollection {
+
+  /**
+   * {@inheritdoc}
+   */
+  protected $pluginKey = 'type';
+
+}

modules/core/data_stream/modules/notification/src/Entity/DataStreamNotification.php

+<?php
+
+namespace Drupal\data_stream_notification\Entity;
+
+use Drupal\Core\Config\Entity\ConfigEntityBase;
+use Drupal\Core\Entity\EntityStorageInterface;
+use Drupal\data_stream_notification\DataStreamNotificationPluginCollection;
+
+/**
+ * Defines the DataStreamNotification entity.
+ *
+ * @ConfigEntityType(
+ *   id = "data_stream_notification",
+ *   label = @Translation("Data stream notification"),
+ *   label_collection = @Translation("Data stream notifications"),
+ *   label_singular = @Translation("data stream notification"),
+ *   label_plural = @Translation("data stream notifications"),
+ *   label_count = @PluralTranslation(
+ *     singular = "@count data stream notification",
+ *     plural = "@count data stream notifications",
+ *   ),
+ *   handlers = {
+ *     "access" = "\Drupal\entity\EntityAccessControlHandler",
+ *     "form" = {
+ *       "add" = "Drupal\data_stream_notification\Form\DataStreamNotificationForm",
+ *       "edit" = "Drupal\data_stream_notification\Form\DataStreamNotificationForm",
+ *       "delete" = "Drupal\Core\Entity\EntityDeleteForm",
+ *     },
+ *     "list_builder" = "Drupal\data_stream_notification\DataStreamNotificationListBuilder",
+ *     "permission_provider" = "\Drupal\entity\EntityPermissionProvider",
+ *     "route_provider" = {
+ *       "default" = "Drupal\entity\Routing\DefaultHtmlRouteProvider",
+ *     },
+ *   },
+ *   admin_permission = "administer data stream notification",
+ *   entity_keys = {
+ *     "id" = "id",
+ *     "label" = "label",
+ *     "status" = "status",
+ *   },
+ *   config_export = {
+ *     "id",
+ *     "label",
+ *     "data_stream",
+ *     "activation_threshold",
+ *     "deactivation_threshold",
+ *     "condition_operator",
+ *     "condition",
+ *     "delivery_interval",
+ *     "delivery",
+ *   },
+ *   links = {
+ *     "collection" = "/data-stream-notifications",
+ *     "add-form" = "/data-stream-notifications/add",
+ *     "edit-form" = "/data-stream-notifications/{data_stream_notification}/edit",
+ *     "delete-form" = "/data-stream-notifications/{data_stream_notification}/delete",
+ *     "enable" = "/data-stream-notifications/{data_stream_notification}/enable",
+ *     "disable" = "/data-stream-notifications/{data_stream_notification}/disable",
+ *   }
+ * )
+ *
+ * @ingroup farm
+ */
+class DataStreamNotification extends ConfigEntityBase implements DataStreamNotificationInterface {
+
+  /**
+   * The notification ID.
+   *
+   * @var string
+   */
+  protected string $id;
+
+  /**
+   * The notification label.
+   *
+   * @var string
+   */
+  protected string $label;
+
+  /**
+   * The data stream ID.
+   *
+   * @var int
+   */
+  protected int $data_stream;
+
+  /**
+   * The activation threshold.
+   *
+   * @var int
+   */
+  protected int $activation_threshold;
+
+  /**
+   * The deactivation threshold.
+   *
+   * @var int
+   */
+  protected int $deactivation_threshold;
+
+  /**
+   * The condition operator.
+   *
+   * @var string
+   */
+  protected string $condition_operator;
+
+  /**
+   * Stores all conditions of this notification.
+   *
+   * @var array
+   */
+  protected array $condition = [];
+
+  /**
+   * The condition plugin collection.
+   *
+   * @var \Drupal\data_stream_notification\DataStreamNotificationPluginCollection
+   */
+  protected $conditionCollection;
+
+  /**
+   * The delivery_interval.
+   *
+   * @var int
+   */
+  protected int $delivery_interval;
+
+  /**
+   * Stores all deliveries of this notification.
+   *
+   * @var array
+   */
+  protected array $delivery = [];
+
+  /**
+   * The delivery plugin collection.
+   *
+   * @var \Drupal\data_stream_notification\DataStreamNotificationPluginCollection
+   */
+  protected $deliveryCollection;
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getPluginCollections() {
+    foreach (['condition', 'delivery'] as $type) {
+      $name = $type . 'Collection';
+      if (empty($this->$name)) {
+        $this->$name = new DataStreamNotificationPluginCollection(\Drupal::service("plugin.manager.data_stream_notification_$type"), $this->$type);
+      }
+    }
+    return [
+      'condition' => $this->conditionCollection,
+      'delivery' => $this->deliveryCollection,
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function postDelete(EntityStorageInterface $storage, array $entities) {
+    parent::postDelete($storage, $entities);
+
+    // Clear the notification state on deletion.
+    foreach ($entities as $entity) {
+      \Drupal::state()->delete($entity->getStateKey());
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function postSave(EntityStorageInterface $storage, $update = TRUE) {
+    parent::postSave($storage, $update);
+
+    // Initialize the notification state on creation.
+    if ($update === FALSE) {
+      $this->resetState();
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getState(): array {
+    return \Drupal::state()->get($this->getStateKey());
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function isActive(): bool {
+    return !empty($this->getState()['active']);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function resetState(bool $active = FALSE): array {
+
+    // Initialize the notification state.
+    $new_state = [
+      'active' => $active,
+      'activate_count' => $active ? 1 : 0,
+      'deactivate_count' => 0,
+    ];
+    \Drupal::state()->set($this->getStateKey(), $new_state);
+    return $new_state;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function incrementState(string $key): array {
+
+    // Bail if an invalid key is provided.
+    if (!in_array($key, ['activate_count', 'deactivate_count'])) {
+      return [];
+    }
+
+    // Get the current state and save a copy as the new_state.
+    $state = \Drupal::state();
+    $notification_key = $this->getStateKey();
+    $current_state = $state->get($notification_key);
+    $new_state = $current_state;
+
+    // Update the desired key in the new_state.
+    $new_state[$key]++;
+
+    // Check if the notification active state should be changed.
+    $current_active_state = $current_state['active'];
+
+    // Reset the other key to 0. This enforces that thresholds are consecutive.
+    // Do not reset the activate_count while the notification is active.
+    $other_key = $key === 'activate_count' ? 'deactivate_count' : 'activate_count';
+    if (!($current_active_state && $other_key === 'activate_count')) {
+      $new_state[$other_key] = 0;
+    }
+
+    // If currently active, check if the deactivation threshold was reached.
+    if ($current_active_state && $new_state['deactivate_count'] >= $this->deactivation_threshold) {
+      return $this->resetState(FALSE);
+    }
+
+    // If not currently active, check if the activation threshold was reached.
+    elseif (!$current_active_state && $new_state['activate_count'] >= $this->activation_threshold) {
+      return $this->resetState(TRUE);
+    }
+
+    // Otherwise just increment the key in the notification state.
+    $state->set($notification_key, $new_state);
+    return $new_state;
+  }
+
+  /**
+   * Helper function to return the state key for the notification.
+   *
+   * @return string
+   *   The state key.
+   */
+  protected function getStateKey() {
+    return 'data_stream_notification.state.' . $this->id();
+  }
+
+}

modules/core/data_stream/modules/notification/src/Entity/DataStreamNotificationInterface.php

+<?php
+
+namespace Drupal\data_stream_notification\Entity;
+
+use Drupal\Core\Config\Entity\ConfigEntityInterface;
+use Drupal\Core\Entity\EntityWithPluginCollectionInterface;
+
+/**
+ * Provides an interface for defining data stream entities.
+ */
+interface DataStreamNotificationInterface extends ConfigEntityInterface, EntityWithPluginCollectionInterface {
+
+  /**
+   * Helper function to get the notification state.
+   *
+   * @return array
+   *   The notification state.
+   */
+  public function getState(): array;
+
+  /**
+   * Helper function to get the notification active state.
+   *
+   * @return bool
+   *   A boolean indicating the notification active state.
+   */
+  public function isActive(): bool;
+
+  /**
+   * Helper function to reset the notification state.
+   *
+   * @param bool $active
+   *   Boolean indicating if the notification is active.
+   *
+   * @return array
+   *   The new notification state.
+   */
+  public function resetState(bool $active): array;
+
+  /**
+   * Helper function to increment the notification state.
+   *
+   * @param string $key
+   *   The state key to set. Either activate_count or deactivate_count.
+   *
+   * @return array
+   *   The new notification state.
+   */
+  public function incrementState(string $key): array;
+
+}

modules/core/data_stream/modules/notification/src/EventSubscriber/DataStreamEventSubscriber.php

+<?php
+
+namespace Drupal\data_stream_notification\EventSubscriber;
+
+use Drupal\Core\Entity\EntityTypeManagerInterface;
+use Drupal\data_stream\Event\DataStreamEvent;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+
+/**
+ * Event subscriber for data stream events.
+ *
+ * Dispatches data stream notifications.
+ */
+class DataStreamEventSubscriber implements EventSubscriberInterface {
+
+  /**
+   * The data stream notification entity storage.
+   *
+   * @var \Drupal\Core\Entity\EntityStorageInterface
+   */
+  protected $notificationStorage;
+
+  /**
+   * Constructs a DataStreamEventSubscriber object.
+   *
+   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager
+   *   The entity type manager service.
+   */
+  public function __construct(EntityTypeManagerInterface $entity_type_manager) {
+    $this->notificationStorage = $entity_type_manager->getStorage('data_stream_notification');
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function getSubscribedEvents() {
+    return [
+      DataStreamEvent::DATA_RECEIVE => 'onDataReceive',
+    ];
+  }
+
+  /**
+   * Trigger notifications when data is received.
+   *
+   * @param \Drupal\data_stream\Event\DataStreamEvent $event
+   *   The data stream event.
+   */
+  public function onDataReceive(DataStreamEvent $event) {
+
+    // Load any notifications configured for the data stream.
+    /** @var \Drupal\data_stream_notification\Entity\DataStreamNotificationInterface[] $notifications */
+    $notifications = $this->notificationStorage->loadByProperties([
+      'status' => TRUE,
+      'data_stream' => $event->dataStream->id(),
+    ]);
+
+    // Bail if there are none.
+    if (empty($notifications)) {
+      return;
+    }
+
+    // Execute all notifications.
+    foreach ($notifications as $notification) {
+      $conditions_met = FALSE;
+
+      // Include the notification in the event context.
+      $event->context['data_stream_notification'] = $notification;
+
+      // Save the configured operator.
+      $operator = $notification->get('condition_operator') ?? 'or';
+
+      // Test each condition plugin and collect the result.
+      $results = [];
+      $summaries = [];
+      $collections = $notification->getPluginCollections();
+      /** @var \Drupal\data_stream_notification\Plugin\DataStream\NotificationCondition\NotificationConditionInterface $condition */
+      foreach ($collections['condition'] as $condition) {
+
+        // Set the event context values on the plugin.
+        $contexts = $condition->getContextDefinitions();
+        foreach ($event->context as $name => $value) {
+          if (array_key_exists($name, $contexts)) {
+            $condition->setContextValue($name, $value);
+          }
+        }
+
+        // Evaluate the condition.
+        $result = $condition->execute();
+
+        // Collect the summary of successful conditions.
+        if ($result) {
+          $summaries[] = $condition->summary();
+        }
+
+        // If success, and the 'or' operator, stop checking conditions.
+        if ($result && $operator === 'or') {
+          $conditions_met = TRUE;
+          break;
+        }
+        $results[] = $result;
+      }
+
+      // Check if the 'and' operator passes.
+      if ($operator === 'and') {
+        $conditions_met = array_product($results);
+      }
+
+      $state_key = $conditions_met ? 'activate_count' : 'deactivate_count';
+      $new_state = $notification->incrementState($state_key);
+
+      // Bail if the notification is not in an active state.
+      if (!$conditions_met || empty($new_state['active'])) {
+        return;
+      }
+
+      // Determine if the notification delivery needs to be executed.
+      // This is based on the notification's configured delivery interval.
+      $execute_delivery = FALSE;
+      $delivery_interval = $notification->get('delivery_interval');
+      $activate_count = $new_state['activate_count'];
+
+      // Always execute delivery when the notification first becomes active.
+      if ($activate_count === 1) {
+        $execute_delivery = TRUE;
+      }
+      // Use modulus arithmetic to determine if the delivery_interval applies.
+      elseif ($delivery_interval > 0 && ($activate_count - 1) % $delivery_interval === 0) {
+        $execute_delivery = TRUE;
+      }
+
+      // Bail if not executing delivery.
+      if (empty($execute_delivery)) {
+        return;
+      }
+
+      // Include the condition summaries.
+      $event->context['condition_summaries'] = $summaries;
+
+      // Else execute all configured delivery plugins.
+      /** @var \Drupal\data_stream_notification\Plugin\DataStream\NotificationDelivery\NotificationDeliveryInterface $delivery */
+      foreach ($collections['delivery'] as $delivery) {
+
+        // Set the event context values on the plugin.
+        $contexts = $delivery->getContextDefinitions();
+        foreach ($event->context as $name => $value) {
+          if (array_key_exists($name, $contexts)) {
+            $delivery->setContextValue($name, $value);
+          }
+        }
+
+        // Execute the delivery plugin.
+        $delivery->execute();
+      }
+    }
+  }
+
+}

modules/core/data_stream/modules/notification/src/Form/DataStreamNotificationForm.php

+<?php
+
+namespace Drupal\data_stream_notification\Form;
+
+use Drupal\Core\Entity\EntityForm;
+use Drupal\Core\Entity\EntityInterface;
+use Drupal\Core\Form\FormStateInterface;
+use Drupal\Core\Form\SubformState;
+use Drupal\data_stream_notification\NotificationConditionManagerInterface;
+use Drupal\data_stream_notification\NotificationDeliveryManagerInterface;
+use Symfony\Component\DependencyInjection\ContainerInterface;
+
+/**
+ * Data stream notification entity form.
+ */
+class DataStreamNotificationForm extends EntityForm {
+
+  /**
+   * The notification condition manager service.
+   *
+   * @var \Drupal\data_stream_notification\NotificationConditionManagerInterface
+   */
+  protected $conditionManager;
+
+  /**
+   * The notification delivery manager service.
+   *
+   * @var \Drupal\data_stream_notification\NotificationDeliveryManagerInterface
+   */
+  protected $deliveryManager;
+
+  /**
+   * Constructs a new DataStreamNotificationForm object.
+   *
+   * @param \Drupal\data_stream_notification\NotificationConditionManagerInterface $condition_manager
+   *   The notification condition manager service.
+   * @param \Drupal\data_stream_notification\NotificationDeliveryManagerInterface $delivery_manager
+   *   The notification delivery manager service.
+   */
+  public function __construct(NotificationConditionManagerInterface $condition_manager, NotificationDeliveryManagerInterface $delivery_manager) {
+    $this->conditionManager = $condition_manager;
+    $this->deliveryManager = $delivery_manager;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public static function create(ContainerInterface $container) {
+    return new static(
+      $container->get('plugin.manager.data_stream_notification_condition'),
+      $container->get('plugin.manager.data_stream_notification_delivery'),
+    );
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function init(FormStateInterface $form_state) {
+    parent::init($form_state);
+
+    // Init the condition and delivery plugins in form_state.
+    $notification = $this->entity;
+    foreach (['condition', 'delivery'] as $plugin_type) {
+      $form_state->setValue($plugin_type, $notification->get($plugin_type));
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function form(array $form, FormStateInterface $form_state) {
+    $form = parent::form($form, $form_state);
+    $notification = $this->entity;
+
+    $form['label'] = [
+      '#type' => 'textfield',
+      '#title' => $this->t('Label'),
+      '#maxlength' => 255,
+      '#default_value' => $notification->label(),
+      '#description' => $this->t('Label for the data stream notification.'),
+      '#required' => TRUE,
+    ];
+
+    $form['id'] = [
+      '#type' => 'machine_name',
+      '#default_value' => $notification->id(),
+      '#machine_name' => [
+        'exists' => '\Drupal\data_stream_notification\Entity\DataStreamNotification::load',
+      ],
+      '#disabled' => !$notification->isNew(),
+    ];
+
+    $form['status'] = [
+      '#type' => 'checkbox',
+      '#title' => $this->t('Enabled'),
+      '#default_value' => $notification->status(),
+    ];
+
+    // @todo Improve data stream entity selection.
+    // @todo Support multiple data streams.
+    $default = $notification->get('data_stream') ?? 0;
+    $form['data_stream'] = [
+      '#type' => 'number',
+      '#title' => $this->t('Data stream ID'),
+      '#required' => TRUE,
+      '#default_value' => $default,
+    ];
+
+    $form['activation_threshold'] = [
+      '#type' => 'number',
+      '#title' => $this->t('Activation threshold'),
+      '#description' => $this->t('How many consecutive times the conditions must be met before notifications are delivered. Defaults to 1, sending a notification the first time conditions are met.'),
+      '#default_value' => $notification->get('activation_threshold') ?? 1,
+      '#min' => 1,
+    ];
+
+    $form['deactivation_threshold'] = [
+      '#type' => 'number',
+      '#title' => $this->t('Deactivation threshold'),
+      '#description' => $this->t('How many consecutive times the conditions must fail before the notification becomes inactive and will require the activation threshold again. Defaults to 1, making the notification inactive the first time conditions are not met.'),
+      '#default_value' => $notification->get('deactivation_threshold') ?? 1,
+      '#min' => 1,
+    ];
+
+    // Define some info about the plugin types.
+    $plugin_types = [
+      'condition' => [
+        'label' => $this->t('Condition'),
+      ],
+      'delivery' => [
+        'label' => $this->t('Delivery'),
+      ],
+    ];
+    foreach ($plugin_types as $plugin_type => $plugin_info) {
+
+      // Get the plugin type manager service.
+      $manager_name = $plugin_type . 'Manager';
+      $manager = $this->$manager_name;
+
+      // Create a wrapper for the plugin type.
+      $wrapper = $plugin_type . '_wrapper';
+      $form[$wrapper] = [
+        '#type' => 'details',
+        '#title' => $plugin_info['label'],
+        '#open' => TRUE,
+        '#prefix' => "<div id=\"$plugin_type-fieldset-wrapper\">",
+        '#suffix' => '</div>',
+      ];
+
+      // Add an actions element.
+      $form[$wrapper][$plugin_type . '_actions'] = [
+        '#type' => 'actions',
+      ];
+
+      // Get the available options for the plugin type.
+      $plugin_options = array_map(function ($definition) {
+        return $definition['label'];
+      }, $manager->getDefinitions());
+      $default = array_keys($plugin_options)[0];
+
+      // Select field for the type of plugin to add.
+      $form[$wrapper][$plugin_type . '_actions'][$plugin_type . '_type'] = [
+        '#type' => 'select',
+        '#title' => $this->t('@type type', ['@type' => $plugin_info['label']]),
+        '#title_display' => 'attribute',
+        '#options' => $plugin_options,
+        '#default_value' => $default,
+      ];
+
+      // Button to add another plugin.
+      $form[$wrapper][$plugin_type . '_actions'][$plugin_type . '_add'] = [
+        '#type' => 'submit',
+        '#value' => $this->t('Add @type', ['@type' => $plugin_info['label']]),
+        '#submit' => ['::addOne'],
+        '#name' => "add-$plugin_type",
+        '#ajax' => [
+          'callback' => '::updatePlugins',
+          'wrapper' => "$plugin_type-fieldset-wrapper",
+        ],
+      ];
+
+      // Start a tree to hold an array of plugin definitions.
+      $form[$wrapper][$plugin_type] = [
+        '#tree' => TRUE,
+      ];
+
+      // Add each plugin definition to the form.
+      $plugins = $form_state->getValue($plugin_type) ?? [];
+      foreach ($plugins as $delta => $plugin_config) {
+
+        // Wrap each plugin definition in a details element.
+        $form[$wrapper][$plugin_type][$delta] = [
+          '#type' => 'details',
+          '#title' => $plugin_options[$plugin_config['type']],
+          '#open' => TRUE,
+        ];
+
+        // Display the plugin type. This can't be changed.
+        $form[$wrapper][$plugin_type][$delta]['type'] = [
+          '#type' => 'hidden',
+          '#title' => $this->t('Condition type'),
+          '#default_value' => $plugin_config['type'],
+          '#disabled' => TRUE,
+        ];
+
+        // Allow the plugin type to provide a subform.
+        $plugin_instance = $manager->createInstance($plugin_config['type'], $plugin_config);
+        $subform_state = SubformState::createForSubform($form[$wrapper][$plugin_type][$delta], $form, $form_state);
+        $form[$wrapper][$plugin_type][$delta] = $plugin_instance->buildConfigurationForm($form[$wrapper][$plugin_type][$delta], $subform_state);
+
+        // Include the summary in condition plugin titles.
+        if ($plugin_type === 'condition' && $summary = $plugin_instance->summary()) {
+          $title = $plugin_options[$plugin_config['type']] . ': ' . $summary;
+          $form[$wrapper][$plugin_type][$delta]['#title'] = $title;
+        }
+
+        // Add button to remove the plugin definition.
+        $form[$wrapper][$plugin_type][$delta]['remove'] = [
+          '#type' => 'submit',
+          '#value' => $this->t('Remove'),
+          '#submit' => ['::removeOne'],
+          '#name' => "remove-$plugin_type-$delta",
+          '#ajax' => [
+            'callback' => '::updatePlugins',
+            'wrapper' => "$plugin_type-fieldset-wrapper",
+          ],
+          '#attributes' => [
+            'class' => ['button--danger'],
+          ],
+          '#weight' => 100,
+        ];
+      }
+    }
+
+    // Condition operator field.
+    $form['condition_wrapper']['condition_operator'] = [
+      '#type' => 'select',
+      '#title' => $this->t('Operator'),
+      '#description' => $this->t('Specify the operator to use when testing conditions. "Or" will trigger the notification when any condition is met; "And" will trigger the notification when all conditions are met.'),
+      '#options' => [
+        'or' => $this->t('Or'),
+        'and' => $this->t('And'),
+      ],
+      '#default_value' => $notification->get('condition_operator') ?? 'or',
+      '#tree' => FALSE,
+      '#weight' => -100,
+    ];
+
+    $form['delivery_wrapper']['delivery_interval'] = [
+      '#type' => 'number',
+      '#title' => $this->t('Delivery interval'),
+      '#description' => $this->t('Once the notification is active, send notifications every N times conditions are met. Defaults to 1, sending a notification every time conditions are met. Set to 10 to send a notification for every 10th incident. Set to 0 to only send a notification when it first becomes active.'),
+      '#default_value' => $notification->get('delivery_interval') ?? 1,
+      '#tree' => FALSE,
+      '#weight' => -100,
+      '#min' => 0,
+    ];
+
+    // Allow the existing notification state to be reset on save.
+    if (!$notification->isNew()) {
+      $current_state = $notification->isActive() ? $this->t('Active') : $this->t('Not active');
+      $title = $this->t('Reset the notification state on save. <em>Current state: @state</em>', ['@state' => $current_state]);
+      $form['reset_state'] = [
+        '#type' => 'checkbox',
+        '#title' => $title,
+        '#default_value' => TRUE,
+        '#weight' => 100,
+      ];
+    }
+
+    return $form;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function validateForm(array &$form, FormStateInterface $form_state) {
+    parent::validateForm($form, $form_state);
+
+    // Check if we are submitting the form.
+    $validate_empty_plugins = FALSE;
+    $parents = $form_state->getTriggeringElement()['#array_parents'];
+    if (count($parents) === 2 && $parents[0] === 'actions' && $parents[1] === 'submit') {
+      $validate_empty_plugins = TRUE;
+    }
+
+    // Allow each condition subform to validate the form.
+    foreach (['condition', 'delivery'] as $plugin_type) {
+      $plugins = $form_state->getValue($plugin_type) ?? [];
+
+      // Validate that at least one plugin is provided when submitting the form.
+      if ($validate_empty_plugins && empty($plugins)) {
+        $form_state->setError($form[$plugin_type . '_wrapper'][$plugin_type . '_actions'][$plugin_type . '_add'], $this->t('At least one @type is required.', ['@type' => $plugin_type]));
+      }
+
+      // Validate each plugin.
+      foreach ($plugins as $delta => $plugin_config) {
+
+        // Get the plugin type manager service.
+        $manager_name = $plugin_type . 'Manager';
+        $manager = $this->$manager_name;
+
+        // Allow the plugin type to validate the subform.
+        $plugin_instance = $manager->createInstance($plugin_config['type'], $plugin_config);
+        $wrapper = $plugin_type . '_wrapper';
+        $subform_state = SubformState::createForSubform($form[$wrapper][$plugin_type][$delta], $form, $form_state);
+        $plugin_instance->validateConfigurationForm($form[$wrapper][$plugin_type][$delta], $subform_state);
+      }
+    }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function submitForm(array &$form, FormStateInterface $form_state) {
+
+    // Allow each plugin subform to submit the form.
+    foreach (['condition', 'delivery'] as $plugin_type) {
+      $plugins = $form_state->getValue($plugin_type) ?? [];
+      foreach ($plugins as $delta => $plugin_config) {
+
+        // Get the plugin type manager service.
+        $manager_name = $plugin_type . 'Manager';
+        $manager = $this->$manager_name;
+
+        // Allow the plugin type to provide a subform.
+        $plugin_instance = $manager->createInstance($plugin_config['type'], $plugin_config);
+        $wrapper = $plugin_type . '_wrapper';
+        $subform_state = SubformState::createForSubform($form[$wrapper][$plugin_type][$delta], $form, $form_state);
+        $plugin_instance->submitConfigurationForm($form[$wrapper][$plugin_type][$delta], $subform_state);
+      }
+    }
+
+    // If specified, reset the notification state.
+    if ($form_state->getValue('reset_state', FALSE)) {
+      $this->entity->resetState();
+    }
+
+    // Call the parent method after subforms are submitted.
+    parent::submitForm($form, $form_state);
+  }
+
+  /**
+   * Submit handler for the "Add {plugin_type}" button.
+   *
+   * Adds additional plugin definitions of the specified type.
+   */
+  public function addOne(array &$form, FormStateInterface $form_state) {
+
+    // Determine the plugin type from the wrapper id, eg: condition_wrapper.
+    $wrapper_id = $form_state->getTriggeringElement()['#array_parents'][0];
+    $plugin_type = substr($wrapper_id, 0, -8);
+
+    // Get existing plugins.
+    $plugins = $form_state->getValue($plugin_type);
+
+    // Add a new plugin of the configured type.
+    $new_plugin_type = $form_state->getValue($plugin_type . '_type');
+    $plugins[] = ['type' => $new_plugin_type, 'settings' => []];
+
+    // Update the form.
+    $form_state->setValue($plugin_type, $plugins);
+    $form_state->setRebuild();
+  }
+
+  /**
+   * Submit handler for the "Remove" button.
+   *
+   * Removes individual plugin definitions.
+   */
+  public function removeOne(array &$form, FormStateInterface $form_state) {
+
+    // Determine which definition to remove.
+    $parents = $form_state->getTriggeringElement()['#array_parents'];
+    $plugin_type = $parents[1];
+    $delta = $parents[2];
+
+    // Remove the delta from the existing plugins.
+    $plugins = $form_state->getValue($plugin_type);
+    unset($plugins[$delta]);
+
+    // Update the form.
+    $form_state->setValue($plugin_type, $plugins);
+    $form_state->setRebuild();
+  }
+
+  /**
+   * Callback for both the "Add" and "Remove" buttons.
+   *
+   * Updates all conditions of the given type.
+   */
+  public function updatePlugins(array &$form, FormStateInterface $form_state) {
+    $wrapper_id = $form_state->getTriggeringElement()['#array_parents'][0];
+    return $form[$wrapper_id];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function save(array $form, FormStateInterface $form_state) {
+    $notification = $this->entity;
+    $status = $notification->save();
+
+    switch ($status) {
+      case SAVED_NEW:
+        $this->messenger()->addMessage($this->t('Created the %label data stream notification.', [
+          '%label' => $notification->label(),
+        ]));
+        break;
+
+      default:
+        $this->messenger()->addMessage($this->t('Saved the %label data stream notification.', [
+          '%label' => $notification->label(),
+        ]));
+    }
+    $form_state->setRedirectUrl($notification->toUrl('collection'));
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function copyFormValuesToEntity(EntityInterface $entity, array $form, FormStateInterface $form_state) {
+    parent::copyFormValuesToEntity($entity, $form, $form_state);
+
+    // Save the condition and delivery values, defaulting to an empty array.
+    // The parent method will skip these since they are plugin collections.
+    $values = $form_state->getValues();
+    foreach (['condition', 'delivery'] as $type) {
+      $entity->set($type, $values[$type] ?? []);
+    }
+  }
+
+}

modules/core/data_stream/modules/notification/src/NotificationConditionManager.php

+<?php
+
+namespace Drupal\data_stream_notification;
+
+use Drupal\Core\Cache\CacheBackendInterface;
+use Drupal\Core\Executable\ExecutableException;
+use Drupal\Core\Executable\ExecutableInterface;
+use Drupal\Core\Extension\ModuleHandlerInterface;
+use Drupal\Core\Plugin\DefaultPluginManager;
+use Drupal\data_stream_notification\Plugin\DataStream\NotificationCondition\NotificationConditionInterface;
+
+/**
+ * Plugin manager for notification condition plugins.
+ */
+class NotificationConditionManager extends DefaultPluginManager implements NotificationConditionManagerInterface {
+
+  /**
+   * Constructs a NotificationConditionManager object.
+   *
+   * @param \Traversable $namespaces
+   *   An object that implements \Traversable which contains the root paths
+   *   keyed by the corresponding namespace to look for plugin implementations.
+   * @param \Drupal\Core\Cache\CacheBackendInterface $cache_backend
+   *   Cache backend instance to use.
+   * @param \Drupal\Core\Extension\ModuleHandlerInterface $module_handler
+   *   The module handler to invoke the alter hook with.
+   */
+  public function __construct(\Traversable $namespaces, CacheBackendInterface $cache_backend, ModuleHandlerInterface $module_handler) {
+    parent::__construct(
+      'Plugin/DataStream/NotificationCondition',
+      $namespaces,
+      $module_handler,
+      'Drupal\data_stream_notification\Plugin\DataStream\NotificationCondition\NotificationConditionInterface',
+      'Drupal\data_stream_notification\Annotation\NotificationCondition',
+    );
+    $this->setCacheBackend($cache_backend, 'data_stream_notification_condition');
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function createInstance($plugin_id, array $configuration = []) {
+    $plugin = parent::createInstance($plugin_id, $configuration);
+
+    // Set the executable manager.
+    return $plugin->setExecutableManager($this);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function execute(ExecutableInterface $plugin) {
+    if ($plugin instanceof NotificationConditionInterface) {
+      $result = $plugin->evaluate();
+      return $plugin->isNegated() ? !$result : $result;
+    }
+    throw new ExecutableException("This manager object can only execute notification condition plugins");
+  }
+
+}

modules/core/data_stream/modules/notification/src/NotificationConditionManagerInterface.php

+<?php
+
+namespace Drupal\data_stream_notification;
+
+use Drupal\Component\Plugin\PluginManagerInterface;
+use Drupal\Core\Executable\ExecutableManagerInterface;
+
+/**
+ * Provides an interface for the notification condition plugin manager.
+ */
+interface NotificationConditionManagerInterface extends PluginManagerInterface, ExecutableManagerInterface {
+
+}