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.client.subscriptions;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.function.Predicate;
  import java.util.stream.Collectors;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.Maps;
  import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
  import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaMonitoredItem;
  import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
  import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
  import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UByte;
  import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
  import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
  import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
  import org.eclipse.milo.opcua.stack.core.types.structured.CreateMonitoredItemsResponse;
  import org.eclipse.milo.opcua.stack.core.types.structured.ModifyMonitoredItemsResponse;
  import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
  import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateResult;
  import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemModifyRequest;
  import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemModifyResult;
  import org.eclipse.milo.opcua.stack.core.types.structured.SetMonitoringModeResponse;
  import org.eclipse.milo.opcua.stack.core.types.structured.SetTriggeringResponse;
  import org.eclipse.milo.opcua.stack.core.util.AsyncSemaphore;
  
  import static com.google.common.collect.Lists.newArrayList;
  import static com.google.common.collect.Lists.newArrayListWithCapacity;
  import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
  import static org.eclipse.milo.opcua.stack.core.util.ConversionUtil.l;
  
  public class OpcUaSubscription implements UaSubscription {
  
      private final Map<UInteger, OpcUaMonitoredItem> itemsByClientHandle = Maps.newConcurrentMap();
      private final Map<UInteger, OpcUaMonitoredItem> itemsByServerHandle = Maps.newConcurrentMap();
  
      private final List<NotificationListener> notificationListeners = new CopyOnWriteArrayList<>();
  
      private final AsyncSemaphore notificationSemaphore = new AsyncSemaphore(1);
  
      private final ClientHandleSequence clientHandleSequence = new ClientHandleSequence(
          itemsByClientHandle::containsKey
      );
  
      final List<UInteger> availableAcknowledgements = Collections.synchronizedList(new ArrayList<>());
  
      private volatile long lastSequenceNumber = 0L;
  
      private volatile double requestedPublishingInterval = 0.0;
      private volatile double revisedPublishingInterval = 0.0;
  
      private volatile UInteger requestedLifetimeCount = uint(0);
      private volatile UInteger revisedLifetimeCount = uint(0);
  
      private volatile UInteger requestedMaxKeepAliveCount = uint(0);
      private volatile UInteger revisedMaxKeepAliveCount = uint(0);
  
      private volatile UInteger maxNotificationsPerPublish;
      private volatile boolean publishingEnabled;
      private volatile UByte priority;
  
      private final OpcUaClient client;
      private final UInteger subscriptionId;
  
      public OpcUaSubscription(
          OpcUaClient client,
          UInteger subscriptionId,
          double revisedPublishingInterval,
          UInteger revisedLifetimeCount,
          UInteger revisedMaxKeepAliveCount,
          UInteger maxNotificationsPerPublish,
          boolean publishingEnabled,
          UByte priority) {
  
          this.client = client;
          this.subscriptionId = subscriptionId;
          this.revisedPublishingInterval = revisedPublishingInterval;
          this.revisedLifetimeCount = revisedLifetimeCount;
          this.revisedMaxKeepAliveCount = revisedMaxKeepAliveCount;
          this.maxNotificationsPerPublish = maxNotificationsPerPublish;
          this.publishingEnabled = publishingEnabled;
          this.priority = priority;
      }
  
      @Override
      public CompletableFuture<List<UaMonitoredItem>> createMonitoredItems(
          TimestampsToReturn timestampsToReturn,
          List<MonitoredItemCreateRequest> itemsToCreate) {
  
          CompletableFuture<CreateMonitoredItemsResponse> future = client.createMonitoredItems(
              subscriptionId,
              timestampsToReturn,
              itemsToCreate
          );
  
          return future.thenApply(response -> {
              List<MonitoredItemCreateResult> results = l(response.getResults());
  
              List<UaMonitoredItem> createdItems = newArrayListWithCapacity(itemsToCreate.size());
  
              for (int i = 0; i < itemsToCreate.size(); i++) {
                  MonitoredItemCreateRequest request = itemsToCreate.get(i);
                  MonitoredItemCreateResult result = results.get(i);
  
                  OpcUaMonitoredItem item = new OpcUaMonitoredItem(
                      client,
                      request.getRequestedParameters().getClientHandle(),
                      request.getItemToMonitor(),
                      result.getMonitoredItemId(),
                      result.getStatusCode(),
                      result.getRevisedSamplingInterval(),
                      result.getRevisedQueueSize(),
                      result.getFilterResult(),
                      request.getMonitoringMode(),
                      request.getRequestedParameters().getFilter(),
                      request.getRequestedParameters().getDiscardOldest(),
                      timestampsToReturn
                  );
  
                  item.setRequestedSamplingInterval(request.getRequestedParameters().getSamplingInterval());
                  item.setRequestedQueueSize(request.getRequestedParameters().getQueueSize());
  
                  if (item.getStatusCode().isGood()) {
                      itemsByClientHandle.put(item.getClientHandle(), item);
                      itemsByServerHandle.put(item.getMonitoredItemId(), item);
                  }
  
                  createdItems.add(item);
              }
  
              return createdItems;
          });
      }
  
      @Override
      public CompletableFuture<List<UaMonitoredItem>> createMonitoredItems(
          TimestampsToReturn timestampsToReturn,
          List<MonitoredItemCreateRequest> itemsToCreate,
          ItemCreationCallback itemCreationCallback) {
  
          return notificationSemaphore.acquire().thenCompose(permit -> {
              CompletableFuture<List<UaMonitoredItem>> itemsFuture =
                  createMonitoredItems(timestampsToReturn, itemsToCreate);
  
              return itemsFuture.whenComplete((items, ex) -> {
                  try {
                      if (items != null) {
                          for (int i = 0; i < items.size(); i++) {
                              UaMonitoredItem item = items.get(i);
  
                              itemCreationCallback.onItemCreated(item, i);
                          }
                      }
                  } finally {
                      permit.release();
                  }
              });
          });
      }
  
      @Override
      public CompletableFuture<List<StatusCode>> modifyMonitoredItems(TimestampsToReturn timestampsToReturn,
                                                                      List<MonitoredItemModifyRequest> itemsToModify) {
  
          CompletableFuture<ModifyMonitoredItemsResponse> future =
              client.modifyMonitoredItems(subscriptionId, timestampsToReturn, itemsToModify);
  
          return future.thenApply(response -> {
              List<StatusCode> statusCodes = newArrayList();
  
              List<MonitoredItemModifyResult> results = l(response.getResults());
  
              for (int i = 0; i < results.size(); i++) {
                  MonitoredItemModifyRequest request = itemsToModify.get(i);
                  MonitoredItemModifyResult result = results.get(i);
                  StatusCode statusCode = result.getStatusCode();
  
                  OpcUaMonitoredItem item = itemsByServerHandle.get(request.getMonitoredItemId());
  
                  if (item != null) {
                      item.setStatusCode(statusCode);
  
                      if (statusCode.isGood()) {
                          item.setTimestamps(timestampsToReturn);
                          item.setRevisedSamplingInterval(result.getRevisedSamplingInterval());
                          item.setRevisedQueueSize(result.getRevisedQueueSize());
                          item.setFilterResult(result.getFilterResult());
                          item.setDiscardOldest(request.getRequestedParameters().getDiscardOldest());
  
                          item.setRequestedFilter(request.getRequestedParameters().getFilter());
                          item.setRequestedSamplingInterval(request.getRequestedParameters().getSamplingInterval());
                          item.setRequestedQueueSize(request.getRequestedParameters().getQueueSize());
                      }
                  }
  
                  statusCodes.add(statusCode);
              }
  
              return statusCodes;
          });
      }
  
      @Override
      public CompletableFuture<List<StatusCode>> deleteMonitoredItems(List<UaMonitoredItem> itemsToDelete) {
  
          List<UInteger> monitoredItemIds = itemsToDelete.stream()
              .map(UaMonitoredItem::getMonitoredItemId)
              .collect(Collectors.toList());
  
          return client.deleteMonitoredItems(subscriptionId, monitoredItemIds).thenApply(response -> {
              List<StatusCode> results = l(response.getResults());
  
              for (int i = 0; i < itemsToDelete.size(); i++) {
                  OpcUaMonitoredItem item = (OpcUaMonitoredItem) itemsToDelete.get(i);
                  itemsByClientHandle.remove(item.getClientHandle());
                  itemsByServerHandle.remove(item.getMonitoredItemId());
                  item.setStatusCode(results.get(i));
              }
  
              return results;
          });
      }
  
  
      @Override
      public CompletableFuture<List<StatusCode>> setMonitoringMode(MonitoringMode monitoringMode,
                                                                   List<UaMonitoredItem> items) {
  
          List<UInteger> monitoredItemIds = items.stream()
              .map(UaMonitoredItem::getMonitoredItemId)
              .collect(Collectors.toList());
  
          CompletableFuture<SetMonitoringModeResponse> future =
              client.setMonitoringMode(subscriptionId, monitoringMode, monitoredItemIds);
  
          return future.thenApply(response -> {
              List<StatusCode> results = l(response.getResults());
  
              for (int i = 0; i < monitoredItemIds.size(); i++) {
                  UInteger id = monitoredItemIds.get(i);
                  OpcUaMonitoredItem item = itemsByServerHandle.get(id);
  
                  StatusCode result = results.get(i);
                  if (result.isGood() && item != null) {
                      item.setMonitoringMode(monitoringMode);
                  }
              }
  
              return results;
          });
      }
  
      @Override
      public CompletableFuture<StatusCode> setPublishingMode(boolean publishingEnabled) {
          return client.setPublishingMode(publishingEnabled, newArrayList(subscriptionId))
              .thenApply(response -> {
                  StatusCode statusCode = l(response.getResults()).get(0);
  
                  if (statusCode.isGood()) {
                      setPublishingEnabled(publishingEnabled);
                  }
  
                  return statusCode;
              });
      }
  
      @Override
      public CompletableFuture<List<StatusCode>> addTriggeringLinks(
          UInteger triggeringItemId,
          List<UInteger> linksToAdd
      ) {
  
          CompletableFuture<SetTriggeringResponse> future = client.setTriggering(
              subscriptionId,
              triggeringItemId,
              linksToAdd,
              Collections.emptyList()
          );
  
          return future.thenApply(r -> Arrays.asList(r.getAddResults()));
      }
  
      @Override
      public CompletableFuture<List<StatusCode>> removeTriggeringLinks(
          UInteger triggeringItemId,
          List<UInteger> linksToRemove
      ) {
  
          CompletableFuture<SetTriggeringResponse> future = client.setTriggering(
              subscriptionId,
              triggeringItemId,
              Collections.emptyList(),
              linksToRemove
          );
  
          return future.thenApply(r -> Arrays.asList(r.getRemoveResults()));
      }
  
      @Override
      public UInteger getSubscriptionId() {
          return subscriptionId;
      }
  
      @Override
      public double getRequestedPublishingInterval() {
          return requestedPublishingInterval;
      }
  
      @Override
      public double getRevisedPublishingInterval() {
          return revisedPublishingInterval;
      }
  
  
      @Override
      public UInteger getRequestedLifetimeCount() {
          return requestedLifetimeCount;
      }
  
      @Override
      public UInteger getRevisedLifetimeCount() {
          return revisedLifetimeCount;
      }
  
      @Override
      public UInteger getRequestedMaxKeepAliveCount() {
          return requestedMaxKeepAliveCount;
      }
  
      @Override
      public UInteger getRevisedMaxKeepAliveCount() {
          return revisedMaxKeepAliveCount;
      }
  
      @Override
      public UInteger getMaxNotificationsPerPublish() {
          return maxNotificationsPerPublish;
      }
  
      @Override
      public boolean isPublishingEnabled() {
          return publishingEnabled;
      }
  
      @Override
      public UByte getPriority() {
          return priority;
      }
  
      @Override
      public ImmutableList<UaMonitoredItem> getMonitoredItems() {
          return ImmutableList.copyOf(itemsByClientHandle.values());
      }
  
      @Override
      public UInteger nextClientHandle() {
          return clientHandleSequence.nextClientHandle();
      }
  
      @Override
      public void addNotificationListener(NotificationListener listener) {
          notificationListeners.add(listener);
      }
  
      @Override
      public void removeNotificationListener(NotificationListener listener) {
          notificationListeners.remove(listener);
      }
  
      List<NotificationListener> getNotificationListeners() {
          return notificationListeners;
      }
  
      AsyncSemaphore getNotificationSemaphore() {
          return notificationSemaphore;
      }
  
      Map<UInteger, OpcUaMonitoredItem> getItemsByClientHandle() {
          return itemsByClientHandle;
      }
  
      Map<UInteger, OpcUaMonitoredItem> getItemsByServerHandle() {
          return itemsByServerHandle;
      }
  
      long getLastSequenceNumber() {
          return lastSequenceNumber;
      }
  
      void setRequestedPublishingInterval(double requestedPublishingInterval) {
          this.requestedPublishingInterval = requestedPublishingInterval;
      }
  
      void setRevisedPublishingInterval(double revisedPublishingInterval) {
          this.revisedPublishingInterval = revisedPublishingInterval;
      }
  
      void setRequestedLifetimeCount(UInteger requestedLifetimeCount) {
          this.requestedLifetimeCount = requestedLifetimeCount;
      }
  
      void setRevisedLifetimeCount(UInteger revisedLifetimeCount) {
          this.revisedLifetimeCount = revisedLifetimeCount;
      }
  
      void setRequestedMaxKeepAliveCount(UInteger requestedMaxKeepAliveCount) {
          this.requestedMaxKeepAliveCount = requestedMaxKeepAliveCount;
      }
  
      void setRevisedMaxKeepAliveCount(UInteger revisedMaxKeepAliveCount) {
          this.revisedMaxKeepAliveCount = revisedMaxKeepAliveCount;
      }
  
      void setMaxNotificationsPerPublish(UInteger maxNotificationsPerPublish) {
          this.maxNotificationsPerPublish = maxNotificationsPerPublish;
      }
  
      void setPublishingEnabled(boolean publishingEnabled) {
          this.publishingEnabled = publishingEnabled;
      }
  
      void setPriority(UByte priority) {
          this.priority = priority;
      }
  
      void setLastSequenceNumber(long lastSequenceNumber) {
          this.lastSequenceNumber = lastSequenceNumber;
      }
  
  
      static class ClientHandleSequence {
  
          private final AtomicLong clientHandle;
  
          private final Predicate<UInteger> handleInUse;
  
          ClientHandleSequence(Predicate<UInteger> handleInUse) {
              this(handleInUse, 0L);
          }
  
          @VisibleForTesting
          ClientHandleSequence(Predicate<UInteger> handleInUse, long initialValue) {
              this.handleInUse = handleInUse;
  
              clientHandle = new AtomicLong(initialValue);
          }
  
          synchronized UInteger nextClientHandle() {
              UInteger original = getAndIncrementWithRollover();
              UInteger next = original;
  
              while (handleInUse.test(next)) {
                  next = getAndIncrementWithRollover();
  
                  if (next.equals(original)) {
                      // All client handles are in use... unlikely!
                      throw new IllegalStateException("no unused client handles");
                  }
              }
  
              return next;
          }
  
          private UInteger getAndIncrementWithRollover() {
              long current = clientHandle.get();
  
              if (current > UInteger.MAX_VALUE) {
                  clientHandle.set(0L);
              }
  
              return uint(clientHandle.getAndIncrement());
          }
  
      }
  
  }
  