/*
 * 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.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;

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.subscriptions.BatchModifyMonitoredItems.ModifyMonitoredItemResult;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemModifyRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.eclipse.milo.opcua.stack.core.util.Unit;

import static java.util.Collections.singletonList;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.eclipse.milo.opcua.stack.core.util.FutureUtils.failedUaFuture;


public class ManagedDataItem extends ManagedItem {

    private final CopyOnWriteArrayList<DataValueListener> dataValueListeners = new CopyOnWriteArrayList<>();

    private UaMonitoredItem.ValueConsumer valueConsumer = null;

    ManagedDataItem(OpcUaClient client, ManagedSubscription subscription, OpcUaMonitoredItem item) {
        super(client, subscription, item);
    }

    @Override
    public CompletableFuture<Unit> deleteAsync() {
        return subscription.deleteDataItemAsync(this);
    }

    //region SamplingInterval operations

    /**
     * Get this item's current sampling interval, i.e. its revised sampling interval.
     *
     * @return this item's current sampling interval, i.e. its revised sampling interval.
     * @see UaMonitoredItem#getRevisedSamplingInterval()
     */
    public double getSamplingInterval() {
        return monitoredItem.getRevisedSamplingInterval();
    }

    /**
     * Request a new sampling interval for this item.
     *
     * @param samplingInterval the new sampling interval to request.
     * @return the new sampling interval, possibly revised by the server.
     * @throws UaException if an operation- or service-level error occurs.
     */
    public double setSamplingInterval(double samplingInterval) throws UaException {
        try {
            return setSamplingIntervalAsync(samplingInterval).get();
        } catch (InterruptedException e) {
            throw new UaException(StatusCodes.Bad_UnexpectedError, e);
        } catch (ExecutionException e) {
            throw UaException.extract(e)
                .orElse(new UaException(StatusCodes.Bad_UnexpectedError, e));
        }
    }

    /**
     * Request a new sampling interval for this item.
     * <p>
     * This call completes asynchronously.
     *
     * @param samplingInterval the new sampling interval to request.
     * @return a {@link CompletableFuture} that completes successfully with the new sampling interval, possibly revised
     * by the server, or completes exceptionally if an operation- or service-level error occurs.
     */
    public CompletableFuture<Double> setSamplingIntervalAsync(double samplingInterval) {
        MonitoringParameters parameters = new MonitoringParameters(
            monitoredItem.getClientHandle(),
            samplingInterval,
            monitoredItem.getMonitoringFilter(),
            // use revised so as not to modify the queue
            // size if it was revised from the original
            monitoredItem.getRevisedQueueSize(),
            monitoredItem.getDiscardOldest()
        );

        MonitoredItemModifyRequest modifyRequest =
            new MonitoredItemModifyRequest(monitoredItem.getMonitoredItemId(), parameters);

        CompletableFuture<List<StatusCode>> future = subscription.getSubscription().modifyMonitoredItems(
            subscription.getDefaultTimestamps(),
            singletonList(modifyRequest)
        );

        return future.thenApply(statusCodes -> statusCodes.get(0)).thenCompose(statusCode -> {
            if (statusCode.isGood()) {
                return completedFuture(monitoredItem.getRevisedSamplingInterval());
            } else {
                return failedUaFuture(statusCode);
            }
        });
    }

    /**
     * Request a new sampling interval for this item as part of a batch operation.
     *
     * @param samplingInterval the new sampling interval to request.
     * @param batch            the {@link BatchModifyMonitoredItems} operation.
     * @return a {@link CompletableFuture} that completes successfully with the new sampling interval, possibly revised
     * by the server, or completes exceptionally if an operation- or service-level error occurred.
     */
    public CompletableFuture<Double> setSamplingIntervalAsync(
        double samplingInterval,
        BatchModifyMonitoredItems batch
    ) {

        CompletableFuture<ModifyMonitoredItemResult> future = batch.add(
            getMonitoredItem(),
            b -> b.setSamplingInterval(samplingInterval)
        );

        return future.thenCompose(result -> {
            if (result.isServiceResultGood()) {
                Optional<CompletableFuture<Double>> opt = result.operationResult().map(s -> {
                    if (s.isGood()) {
                        return completedFuture(getSamplingInterval());
                    } else {
                        return failedUaFuture(s);
                    }
                });

                // if the service result is good the operation result must be present.
                return opt.orElse(failedUaFuture(new StatusCode(StatusCodes.Bad_InternalError)));
            } else {
                return failedUaFuture(result.serviceResult());
            }
        });
    }

    //endregion

    //region DataValueListener bookkeeping

    /**
     * Add a {@link DataValue} {@link Consumer} to this {@link ManagedDataItem}.
     * <p>
     * {@code consumer} will be invoked any time a new {@link DataValue} arrives for this item.
     * <p>
     * The Consumer is transformed into the returned {@link DataValueListener} that can later be removed.
     * <p>
     * Listeners should be added in the callback that happens during item creation in the
     * {@link ManagedSubscription#createDataItem(double, ReadValueId, Consumer)} and
     * {@link ManagedSubscription#createDataItemsAsync(double, List, Consumer)} to avoid a possible
     * race condition between the initial value arriving and the listener being registered.
     *
     * @param consumer a {@link DataValue} {@link Consumer}.
     * @return a {@link DataValueListener} that can later be removed.
     */
    public synchronized DataValueListener addDataValueListener(Consumer<DataValue> consumer) {
        DataValueListener dataValueListener = (item, value) -> consumer.accept(value);

        addDataValueListener(dataValueListener);

        return dataValueListener;
    }

    /**
     * Add a {@link DataValueListener} to this {@link ManagedDataItem}.
     * <p>
     * {@code listener} will be invoked any time a new {@link DataValue} arrives for this item.
     * <p>
     * Listeners should be added in the callback that happens during item creation in the
     * {@link ManagedSubscription#createDataItem(double, ReadValueId, Consumer)} and
     * {@link ManagedSubscription#createDataItemsAsync(double, List, Consumer)} to avoid a possible
     * race condition between the initial value arriving and the listener being registered.
     *
     * @param dataValueListener the {@link DataValueListener} to add.
     */
    public synchronized void addDataValueListener(DataValueListener dataValueListener) {
        dataValueListeners.add(dataValueListener);

        if (valueConsumer == null) {
            valueConsumer = new ManagedValueConsumer();
            monitoredItem.setValueConsumer(valueConsumer);
        }
    }

    /**
     * Remove a {@link DataValueListener} from this {@link ManagedDataItem}.
     *
     * @param dataValueListener the {@link DataValueListener} to remove.
     * @return {@code true} if the listener was removed.
     */
    public synchronized boolean removeDataValueListener(DataValueListener dataValueListener) {
        boolean removed = dataValueListeners.remove(dataValueListener);

        if (dataValueListeners.isEmpty()) {
            monitoredItem.setValueConsumer((UaMonitoredItem.ValueConsumer) null);
            valueConsumer = null;
        }

        return removed;
    }

    //endregion

    /**
     * A callback that receives notification of new values for a {@link ManagedDataItem}.
     */
    public interface DataValueListener {

        /**
         * A new {@link DataValue} for {@code item} has arrived.
         * <p>
         * Take care not to block unnecessarily in this callback because subscription notifications are processed
         * synchronously as a backpressure mechanism. Blocking inside this callback will prevent subsequent
         * notifications from being processed and new PublishRequests from being sent.
         *
         * @param item  the {@link ManagedDataItem} for which a new value has arrived.
         * @param value the new {@link DataValue}.
         */
        void onDataValueReceived(ManagedDataItem item, DataValue value);

    }

    private class ManagedValueConsumer implements UaMonitoredItem.ValueConsumer {
        @Override
        public void onValueArrived(UaMonitoredItem item, DataValue value) {
            dataValueListeners.forEach(
                dataValueListener ->
                    dataValueListener.onDataValueReceived(ManagedDataItem.this, value)
            );
        }
    }

}
