code: /*
   * Copyright (c) 2019 the Eclipse Milo Authors
   *
   * This program and the accompanying materials are made
   * available under the terms of the Eclipse Public License 2.0
   * which is available at https://www.eclipse.org/legal/epl-2.0/
   *
   * SPDX-License-Identifier: EPL-2.0
   */
  
  package org.eclipse.milo.opcua.sdk.server.util;
  
  import java.math.RoundingMode;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  import com.google.common.math.DoubleMath;
  import org.eclipse.milo.opcua.sdk.core.util.GroupMapCollate;
  import org.eclipse.milo.opcua.sdk.server.AbstractLifecycle;
  import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
  import org.eclipse.milo.opcua.sdk.server.api.DataItem;
  import org.eclipse.milo.opcua.sdk.server.api.MonitoredItem;
  import org.eclipse.milo.opcua.sdk.server.api.services.AttributeServices;
  import org.eclipse.milo.opcua.sdk.server.api.services.AttributeServices.ReadContext;
  import org.eclipse.milo.opcua.stack.core.AttributeId;
  import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
  import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
  import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
  import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
  import org.eclipse.milo.opcua.stack.core.util.ExecutionQueue;
  
  public class SubscriptionModel extends AbstractLifecycle {
  
      private final Set<DataItem> itemSet = Collections.newSetFromMap(Maps.newConcurrentMap());
  
      private final List<ScheduledUpdate> schedule = Lists.newCopyOnWriteArrayList();
  
      private final ExecutorService executor;
      private final ScheduledExecutorService scheduler;
      private final ExecutionQueue executionQueue;
  
      private final OpcUaServer server;
      private final AttributeServices attributeServices;
  
      public SubscriptionModel(OpcUaServer server, AttributeServices attributeServices) {
          this.server = server;
  
          this.attributeServices = attributeServices;
  
          executor = server.getExecutorService();
          scheduler = server.getScheduledExecutorService();
  
          executionQueue = new ExecutionQueue(executor);
      }
  
      @Override
      protected void onStartup() {
      }
  
      @Override
      protected void onShutdown() {
          executionQueue.submit(() -> {
              schedule.forEach(ScheduledUpdate::cancel);
              schedule.clear();
              itemSet.clear();
          });
      }
  
      public void onDataItemsCreated(List<DataItem> items) {
          if (isNotRunning()) {
              throw new IllegalArgumentException("not running");
          }
  
          executionQueue.submit(() -> {
              itemSet.addAll(items);
              reschedule();
          });
      }
  
      public void onDataItemsModified(List<DataItem> items) {
          if (isNotRunning()) {
              throw new IllegalArgumentException("not running");
          }
  
          executionQueue.submit(this::reschedule);
      }
  
      public void onDataItemsDeleted(List<DataItem> items) {
          if (isNotRunning()) {
              throw new IllegalArgumentException("not running");
          }
  
          executionQueue.submit(() -> {
              itemSet.removeAll(items);
              reschedule();
          });
      }
  
      public void onMonitoringModeChanged(List<MonitoredItem> items) {
          if (isNotRunning()) {
              throw new IllegalArgumentException("not running");
          }
  
          executionQueue.submit(this::reschedule);
      }
  
      private void reschedule() {
          Map<Double, List<DataItem>> bySamplingInterval = itemSet.stream()
              .filter(DataItem::isSamplingEnabled)
              .collect(Collectors.groupingBy(DataItem::getSamplingInterval));
  
          List<ScheduledUpdate> updates = bySamplingInterval.keySet().stream()
              .map(samplingInterval -> {
                  List<DataItem> items = bySamplingInterval.get(samplingInterval);
  
                  return new ScheduledUpdate(samplingInterval, items);
              })
              .collect(Collectors.toList());
  
          schedule.forEach(ScheduledUpdate::cancel);
          schedule.clear();
          schedule.addAll(updates);
          schedule.forEach(scheduler::execute);
      }
  
      private class ScheduledUpdate implements Runnable {
  
          private volatile boolean cancelled = false;
  
          private final long samplingInterval;
          private final List<DataItem> items;
  
          private ScheduledUpdate(double samplingInterval, List<DataItem> items) {
              this.samplingInterval = DoubleMath.roundToLong(samplingInterval, RoundingMode.UP);
              this.items = items;
          }
  
          private void cancel() {
              cancelled = true;
          }
  
          @Override
          public void run() {
              if (cancelled) return;
  
              CompletableFuture<List<DataValue>> future = GroupMapCollate.groupMapCollate(
                  items,
                  MonitoredItem::getSession,
                  session -> sessionItems -> {
                      List<PendingRead> pending = sessionItems.stream()
                          .map(item -> new PendingRead(item.getReadValueId()))
                          .collect(Collectors.toList());
  
                      List<ReadValueId> ids = pending.stream()
                          .map(PendingRead::getInput)
                          .collect(Collectors.toList());
  
                      ReadContext context = new ReadContext(server, session);
  
                      attributeServices.read(context, 0d, TimestampsToReturn.Both, ids);
  
                      return context.getFuture();
                  }
              );
  
              future.thenAcceptAsync(values -> {
                  Iterator<DataItem> ii = items.iterator();
                  Iterator<DataValue> vi = values.iterator();
  
                  while (ii.hasNext() && vi.hasNext()) {
                      DataItem item = ii.next();
                      DataValue value = vi.next();
  
                      TimestampsToReturn timestamps = item.getTimestampsToReturn();
  
                      if (timestamps != null) {
                          UInteger attributeId = item.getReadValueId().getAttributeId();
  
                          value = (AttributeId.Value.isEqual(attributeId)) ?
                              DataValue.derivedValue(value, timestamps) :
                              DataValue.derivedNonValue(value, timestamps);
                      }
  
                      item.setValue(value);
                  }
  
                  if (!cancelled) {
                      scheduler.schedule(this, samplingInterval, TimeUnit.MILLISECONDS);
                  }
              }, executor);
          }
  
      }
  
  }
  