<commit_before>/**
 * Copyright (C) 2010-12 Brookhaven National Laboratory
 * All rights reserved. Use is subject to license terms.
 */
package org.epics.pvmanager;

import java.lang.ref.WeakReference;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.epics.pvmanager.expression.DesiredRateExpression;
import org.epics.util.time.TimeDuration;

/**
 * Object responsible to notify the PVReader of changes on the appropriate thread.
 *
 * @author carcassi
 */
public class PVReaderDirector<T> {
    
    private static final Logger log = Logger.getLogger(PVReaderDirector.class.getName());

    // Required for connection and exception notification

    /** Executor used to notify of new values/connection/exception */
    private final Executor notificationExecutor;
    /** Executor used to scan the connection/exception queues */
    private final ScheduledExecutorService scannerExecutor;
    private volatile ScheduledFuture<?> scanTaskHandle;
    /** PVReader to update during the notification */
    private final WeakReference<PVReaderImpl<T>> pvRef;
    /** Function for the new value */
    private final ReadFunction<T> function;
    
    // Required to connect/disconnect expressions
    private final DataSource dataSource;
    private final Object lock = new Object();
    private final Map<DesiredRateExpression<?>, ReadRecipe> recipes =
            new HashMap<>();

    // Required for multiple operations
    /** Connection collector required to connect/disconnect expressions and for connection notification */
    private final ConnectionCollector connCollector =
            new ConnectionCollector();
    /** Exception queue to be used to connect/disconnect expression and for exception notification */
    private final QueueCollector<Exception> exceptionCollector =
            new QueueCollector<>(1);
    
    
    ReadRecipe getCurrentReadRecipe() {
        ReadRecipeBuilder builder = new ReadRecipeBuilder();
        for (Map.Entry<DesiredRateExpression<?>, ReadRecipe> entry : recipes.entrySet()) {
            ReadRecipe readRecipe = entry.getValue();
            for (ChannelReadRecipe channelReadRecipe : readRecipe.getChannelReadRecipes()) {
                builder.addChannel(channelReadRecipe.getChannelName(), channelReadRecipe.getReadSubscription().getValueCache());
            }
        }
        return builder.build(exceptionCollector, connCollector);
    }
    
    /**
     * Calculate the recipes and connects the channel to the datasource.
     * 
     * @param expression 
     */
    public void connectExpression(DesiredRateExpression<?> expression) {
        ReadRecipeBuilder builder = new ReadRecipeBuilder();
        expression.fillReadRecipe(this, builder);
        ReadRecipe recipe = builder.build(exceptionCollector, connCollector);
        synchronized(lock) {
            recipes.put(expression, recipe);
        }
        if (!recipe.getChannelReadRecipes().isEmpty()) {
            try {
                dataSource.connectRead(recipe);
            } catch(Exception ex) {
                recipe.getChannelReadRecipes().iterator().next().getReadSubscription().getExceptionWriteFunction().writeValue(ex);
            }
        }
    }
    
    public void disconnectExpression(DesiredRateExpression<?> expression) {
        ReadRecipe recipe;
        synchronized(lock) {
            recipe = recipes.remove(expression);
        }
        if (recipe == null) {
            log.log(Level.SEVERE, "Director was asked to disconnect expression '" + expression + "' which was not found.");
        }
        
        if (!recipe.getChannelReadRecipes().isEmpty()) {
            try {
                dataSource.disconnectRead(recipe);
            } catch(Exception ex) {
                recipe.getChannelReadRecipes().iterator().next().getReadSubscription().getExceptionWriteFunction().writeValue(ex);
            }
        }
    }
    
    public void close() {
        synchronized(lock) {
            while (!recipes.isEmpty()) {
                DesiredRateExpression<?> expression = recipes.keySet().iterator().next();
                disconnectExpression(expression);
            }
        }
    }

    /**
     * Creates a new notifier. The new notifier will notifier the given pv
     * with new values calculated by the function, and will use onThread to
     * perform the notifications.
     * <p>
     * After construction, one MUST set the pvRecipe, so that the
     * dataSource is appropriately closed.
     *
     * @param pv the pv on which to notify
     * @param function the function used to calculate new values
     * @param notificationExecutor the thread switching mechanism
     */
    PVReaderDirector(PVReaderImpl<T> pv, ReadFunction<T> function, ScheduledExecutorService scannerExecutor,
            Executor notificationExecutor, DataSource dataSource) {
        this.pvRef = new WeakReference<>(pv);
        this.function = function;
        this.notificationExecutor = notificationExecutor;
        this.scannerExecutor = scannerExecutor;
        this.dataSource = dataSource;
    }

    /**
     * Determines whether the notifier is active or not.
     * <p>
     * The notifier becomes inactive if the PVReader is closed or is garbage collected.
     * The first time this function determines that the notifier is inactive,
     * it will ask the data source to close all channels relative to the
     * pv.
     *
     * @return true if new notification should be performed
     */
    boolean isActive() {
        // Making sure to get the reference once for thread safety
        final PVReader<T> pv = pvRef.get();
        if (pv != null && !pv.isClosed()) {
            return true;
        } else {
            return false;
        }
    }
    
    /**
     * Checks whether the pv is paused
     * 
     * @return true if paused
     */
    boolean isPaused() {
        final PVReader<T> pv = pvRef.get();
        if (pv == null || pv.isPaused()) {
            return true;
        } else {
            return false;
        }
    }
    
    private volatile boolean notificationInFlight = false;
    
    /**
     * Notifies the PVReader of a new value.
     */
    void notifyPv() {
        // Don't even calculate if notification is in flight.
        // This makes pvManager automatically throttle back if the consumer
        // is slower than the producer.
        if (notificationInFlight)
            return;<commit_msg>Javadoc update.<commit_after>/**
 * Copyright (C) 2010-12 Brookhaven National Laboratory
 * All rights reserved. Use is subject to license terms.
 */
package org.epics.pvmanager;

import java.lang.ref.WeakReference;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.epics.pvmanager.expression.DesiredRateExpression;
import org.epics.util.time.TimeDuration;

/**
 * Object responsible to notify the PVReader of changes on the appropriate thread.
 * <p>
 * This class is responsible for the correct read operation, including:
 * <ul>
 * <li>Setting up the collector for notifications</li>
 * <li>Setting up the collector for connection notification</li>
 * <li>Building connection recipes and forwarding them to the datasource<li>
 * <li>Managing the scanning task and notification for new values, connection status
 * or errors</li>
 * <li>Disconnecting the expressions from the datasources if the reader is closed
 * or if it's garbage collected</li>
 * </ul>
 *
 * @param <T> value type for the reader managed by this director
 * @author carcassi
 */
public class PVReaderDirector<T> {
    
    private static final Logger log = Logger.getLogger(PVReaderDirector.class.getName());

    // Required for connection and exception notification

    /** Executor used to notify of new values/connection/exception */
    private final Executor notificationExecutor;
    /** Executor used to scan the connection/exception queues */
    private final ScheduledExecutorService scannerExecutor;
    private volatile ScheduledFuture<?> scanTaskHandle;
    /** PVReader to update during the notification */
    private final WeakReference<PVReaderImpl<T>> pvRef;
    /** Function for the new value */
    private final ReadFunction<T> function;
    
    // Required to connect/disconnect expressions
    private final DataSource dataSource;
    private final Object lock = new Object();
    private final Map<DesiredRateExpression<?>, ReadRecipe> recipes =
            new HashMap<>();

    // Required for multiple operations
    /** Connection collector required to connect/disconnect expressions and for connection notification */
    private final ConnectionCollector connCollector =
            new ConnectionCollector();
    /** Exception queue to be used to connect/disconnect expression and for exception notification */
    private final QueueCollector<Exception> exceptionCollector =
            new QueueCollector<>(1);
    
    
    ReadRecipe getCurrentReadRecipe() {
        ReadRecipeBuilder builder = new ReadRecipeBuilder();
        for (Map.Entry<DesiredRateExpression<?>, ReadRecipe> entry : recipes.entrySet()) {
            ReadRecipe readRecipe = entry.getValue();
            for (ChannelReadRecipe channelReadRecipe : readRecipe.getChannelReadRecipes()) {
                builder.addChannel(channelReadRecipe.getChannelName(), channelReadRecipe.getReadSubscription().getValueCache());
            }
        }
        return builder.build(exceptionCollector, connCollector);
    }
    
    /**
     * Connects the given expression.
     * <p>
     * This can be used for dynamic expression to add and connect child expressions.
     * The added expression will be automatically closed when the associated
     * reader is closed, if it's not disconnected first.
     * 
     * @param expression the expression to connect
     */
    public void connectExpression(DesiredRateExpression<?> expression) {
        ReadRecipeBuilder builder = new ReadRecipeBuilder();
        expression.fillReadRecipe(this, builder);
        ReadRecipe recipe = builder.build(exceptionCollector, connCollector);
        synchronized(lock) {
            recipes.put(expression, recipe);
        }
        if (!recipe.getChannelReadRecipes().isEmpty()) {
            try {
                dataSource.connectRead(recipe);
            } catch(Exception ex) {
                recipe.getChannelReadRecipes().iterator().next().getReadSubscription().getExceptionWriteFunction().writeValue(ex);
            }
        }
    }
    
    /**
     * Disconnects the given expression.
     * <p>
     * This can be used for dynamic expression, to remove and disconnects child
     * expressions.
     *
     * @param expression the expression to disconnect
     */
    public void disconnectExpression(DesiredRateExpression<?> expression) {
        ReadRecipe recipe;
        synchronized(lock) {
            recipe = recipes.remove(expression);
        }
        if (recipe == null) {
            log.log(Level.SEVERE, "Director was asked to disconnect expression '" + expression + "' which was not found.");
        }
        
        if (!recipe.getChannelReadRecipes().isEmpty()) {
            try {
                dataSource.disconnectRead(recipe);
            } catch(Exception ex) {
                recipe.getChannelReadRecipes().iterator().next().getReadSubscription().getExceptionWriteFunction().writeValue(ex);
            }
        }
    }

    /**
     * Closed and disconnects all the child expressions.
     */
    private void close() {
        synchronized(lock) {
            while (!recipes.isEmpty()) {
                DesiredRateExpression<?> expression = recipes.keySet().iterator().next();
                disconnectExpression(expression);
            }
        }
    }

    /**
     * Creates a new notifier. The new notifier will notifier the given pv
     * with new values calculated by the function, and will use onThread to
     * perform the notifications.
     * <p>
     * After construction, one MUST set the pvRecipe, so that the
     * dataSource is appropriately closed.
     *
     * @param pv the pv on which to notify
     * @param function the function used to calculate new values
     * @param notificationExecutor the thread switching mechanism
     */
    PVReaderDirector(PVReaderImpl<T> pv, ReadFunction<T> function, ScheduledExecutorService scannerExecutor,
            Executor notificationExecutor, DataSource dataSource) {
        this.pvRef = new WeakReference<>(pv);
        this.function = function;
        this.notificationExecutor = notificationExecutor;
        this.scannerExecutor = scannerExecutor;
        this.dataSource = dataSource;
    }

    /**
     * Determines whether the notifier is active or not.
     * <p>
     * The notifier becomes inactive if the PVReader is closed or is garbage collected.
     * The first time this function determines that the notifier is inactive,
     * it will ask the data source to close all channels relative to the
     * pv.
     *
     * @return true if new notification should be performed
     */
    boolean isActive() {
        // Making sure to get the reference once for thread safety
        final PVReader<T> pv = pvRef.get();
        if (pv != null && !pv.isClosed()) {
            return true;
        } else if (pv == null) {
            log.warning("PVReader wasn't properly closed and it was garbage collected. Closing the associated connections...");
            return false;
        } else {
            return false;
        }
    }
    
    /**
     * Checks whether the pv is paused
     * 
     * @return true if paused
     */
    boolean isPaused() {
        final PVReader<T> pv = pvRef.get();
        if (pv == null || pv.isPaused()) {
            return true;
        } else {
            return false;
        }
    }
    
    private volatile boolean notificationInFlight = false;
    
    /**
     * Notifies the PVReader of a new value.
     */
    void notifyPv() {
        // Don't even calculate if notification is in flight.
        // This makes pvManager automatically throttle back if the consumer
        // is slower than the producer.
        if (notificationInFlight)
            return;
{'fraction_non_alphanumeric': 0.08661578301758212, 'fraction_numerical': 0.000954068420335287, 'mean_word_length': 2.606537232735316, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2142805', 'n_tokens_mistral': 3734, 'n_tokens_neox': 3632, 'n_words': 1165}