<commit_before>/*
 * Copyright (c) 2011 Stiftung Deutsches Elektronen-Synchrotron,
 * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY.
 *
 * THIS SOFTWARE IS PROVIDED UNDER THIS LICENSE ON AN "../AS IS" BASIS.
 * WITHOUT WARRANTY OF ANY KIND, EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED
 * TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR PARTICULAR PURPOSE AND
 * NON-INFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
 * FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR
 * THE USE OR OTHER DEALINGS IN THE SOFTWARE. SHOULD THE SOFTWARE PROVE DEFECTIVE
 * IN ANY RESPECT, THE USER ASSUMES THE COST OF ANY NECESSARY SERVICING, REPAIR OR
 * CORRECTION. THIS DISCLAIMER OF WARRANTY CONSTITUTES AN ESSENTIAL PART OF THIS LICENSE.
 * NO USE OF ANY SOFTWARE IS AUTHORIZED HEREUNDER EXCEPT UNDER THIS DISCLAIMER.
 * DESY HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS,
 * OR MODIFICATIONS.
 * THE FULL LICENSE SPECIFYING FOR THE SOFTWARE THE REDISTRIBUTION, MODIFICATION,
 * USAGE AND OTHER RIGHTS AND OBLIGATIONS IS INCLUDED WITH THE DISTRIBUTION OF THIS
 * PROJECT IN THE FILE LICENSE.HTML. IF THE LICENSE IS NOT INCLUDED YOU MAY FIND A COPY
 * AT HTTP://WWW.DESY.DE/LEGAL/LICENSE.HTM
 */
package org.csstudio.archive.common.engine.pvmanager;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.csstudio.archive.common.engine.model.ArchiveChannelBuffer;
import org.csstudio.archive.common.engine.model.ArchiveEngineSampleRescuer;
import org.csstudio.archive.common.engine.model.EngineModelException;
import org.csstudio.archive.common.engine.service.IServiceProvider;
import org.csstudio.archive.common.service.ArchiveServiceException;
import org.csstudio.archive.common.service.IArchiveEngineFacade;
import org.csstudio.archive.common.service.channel.ArchiveChannelId;
import org.csstudio.archive.common.service.sample.ArchiveMultiScalarSample;
import org.csstudio.archive.common.service.sample.ArchiveSample;
import org.csstudio.archive.common.service.sample.IArchiveSample;
import org.csstudio.archive.common.service.util.ArchiveTypeConversionSupport;
import org.csstudio.domain.desy.epics.types.EpicsGraphicsData;
import org.csstudio.domain.desy.epics.types.EpicsMetaData;
import org.csstudio.domain.desy.epics.types.EpicsSystemVariable;
import org.csstudio.domain.desy.service.osgi.OsgiServiceUnavailableException;
import org.csstudio.domain.desy.system.ISystemVariable;
import org.epics.pvmanager.PVReader;
import org.epics.pvmanager.PVReaderListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Listener for creating archive samples per value update.
 *
 * @author bknerr
 * @since 30.08.2011
 * @param <V> the basic element type of the variable's value collection
 * @param <T> the generic system variable type
 */
// CHECKSTYLE OFF: AbstractClassName
public abstract class DesyArchivePVManagerListener<V extends Serializable,
                                                   T extends ISystemVariable<V>> implements PVReaderListener {
// CHECKSTYLE ON: AbstractClassName

    private static final Logger LOG = LoggerFactory.getLogger(DesyArchivePVManagerListener.class);
    private static final Logger EMAIL_LOG = LoggerFactory.getLogger("ErrorPerEmailLogger");

    private final ArchiveChannelBuffer<V, T> _buffer;
    private IServiceProvider _provider;
    private final String _channelName;
    private final ArchiveChannelId _channelId;
    private String _datatype;
    private volatile boolean _firstConnection;

    private String _startInfo;
    @SuppressWarnings("unused")
    private String _stopInfo;

    private final PVReader<?> _reader;
    /**
     * Constructor.
     */
    public DesyArchivePVManagerListener(@Nonnull final ArchiveChannelBuffer<V, T> buffer,
                                        @Nonnull final PVReader<?> reader,
                                        @Nonnull final IServiceProvider provider,
                                        @Nonnull final String name,
                                        @Nonnull final ArchiveChannelId id,
                                        @Nullable final String datatype) {
        _buffer = buffer;
        _reader = reader;
        _provider = provider;
        _channelName = name;
        _channelId = id;
        _datatype = datatype;
        _firstConnection = true;
    }


    /**
     * {@inheritDoc}
     */
    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void pvChanged() {
        try {
            // TODO (bknerr) : ask Gabriele whether it is a good choice to separate the value
            // update callback from the calling/value providing instance

            final List<EpicsSystemVariable> sysVars = (List<EpicsSystemVariable>) _reader.getValue();
            if (_firstConnection && !sysVars.isEmpty()) {
                handleOnConnectionInformation(_provider, sysVars.get(0), _channelId, _buffer.isConnected(), _startInfo);
                _firstConnection = false;
            }
            for (final EpicsSystemVariable sysVar : sysVars) {
                handleValueUpdateInformation(sysVar);
            }
        } catch (final Throwable t) {
            LOG.error("Unexpected exception in PVListener for: {}:\n{}", _channelName, t.getMessage());
        }
    }

    @SuppressWarnings("rawtypes")
    private void handleOnConnectionInformation(@Nonnull final IServiceProvider provider,
                                               @Nonnull final EpicsSystemVariable pv,
                                               @Nonnull final ArchiveChannelId id,
                                               final boolean connected,
                                               @Nonnull final String info)
                                               throws EngineModelException, OsgiServiceUnavailableException, ArchiveServiceException {
        _buffer.persistChannelStatusInfo(id, connected, info);

        if(!validateAndPersistDatatype(provider, id, pv.getData())) {
            _buffer.stop("Datatype mismatch");
            return;
        }

        final EpicsMetaData metaData = pv.getMetaData();
        if (metaData != null) {
            handleMetaDataInfo(id, metaData);
        }
    }

    private boolean validateAndPersistDatatype(@Nonnull final IServiceProvider provider,
                                               @Nonnull final ArchiveChannelId id,
                                               @Nonnull final Object data) throws OsgiServiceUnavailableException,
                                                                                     ArchiveServiceException {

        final String actualTypeFromData = ArchiveTypeConversionSupport.createArchiveTypeStringFromData(data);
<commit_msg>o.c.archive.common.engine: write number of samples received my pvmanager<commit_after>/*
 * Copyright (c) 2011 Stiftung Deutsches Elektronen-Synchrotron,
 * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY.
 *
 * THIS SOFTWARE IS PROVIDED UNDER THIS LICENSE ON AN "../AS IS" BASIS.
 * WITHOUT WARRANTY OF ANY KIND, EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED
 * TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR PARTICULAR PURPOSE AND
 * NON-INFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE
 * FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR
 * THE USE OR OTHER DEALINGS IN THE SOFTWARE. SHOULD THE SOFTWARE PROVE DEFECTIVE
 * IN ANY RESPECT, THE USER ASSUMES THE COST OF ANY NECESSARY SERVICING, REPAIR OR
 * CORRECTION. THIS DISCLAIMER OF WARRANTY CONSTITUTES AN ESSENTIAL PART OF THIS LICENSE.
 * NO USE OF ANY SOFTWARE IS AUTHORIZED HEREUNDER EXCEPT UNDER THIS DISCLAIMER.
 * DESY HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS,
 * OR MODIFICATIONS.
 * THE FULL LICENSE SPECIFYING FOR THE SOFTWARE THE REDISTRIBUTION, MODIFICATION,
 * USAGE AND OTHER RIGHTS AND OBLIGATIONS IS INCLUDED WITH THE DISTRIBUTION OF THIS
 * PROJECT IN THE FILE LICENSE.HTML. IF THE LICENSE IS NOT INCLUDED YOU MAY FIND A COPY
 * AT HTTP://WWW.DESY.DE/LEGAL/LICENSE.HTM
 */
package org.csstudio.archive.common.engine.pvmanager;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.csstudio.archive.common.engine.model.ArchiveChannelBuffer;
import org.csstudio.archive.common.engine.model.ArchiveEngineSampleRescuer;
import org.csstudio.archive.common.engine.model.EngineModelException;
import org.csstudio.archive.common.engine.service.IServiceProvider;
import org.csstudio.archive.common.service.ArchiveServiceException;
import org.csstudio.archive.common.service.IArchiveEngineFacade;
import org.csstudio.archive.common.service.channel.ArchiveChannelId;
import org.csstudio.archive.common.service.sample.ArchiveMultiScalarSample;
import org.csstudio.archive.common.service.sample.ArchiveSample;
import org.csstudio.archive.common.service.sample.IArchiveSample;
import org.csstudio.archive.common.service.util.ArchiveTypeConversionSupport;
import org.csstudio.domain.desy.epics.types.EpicsGraphicsData;
import org.csstudio.domain.desy.epics.types.EpicsMetaData;
import org.csstudio.domain.desy.epics.types.EpicsSystemVariable;
import org.csstudio.domain.desy.service.osgi.OsgiServiceUnavailableException;
import org.csstudio.domain.desy.system.ISystemVariable;
import org.epics.pvmanager.PVReader;
import org.epics.pvmanager.PVReaderListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Listener for creating archive samples per value update.
 *
 * @author bknerr
 * @since 30.08.2011
 * @param <V> the basic element type of the variable's value collection
 * @param <T> the generic system variable type
 */
// CHECKSTYLE OFF: AbstractClassName
public abstract class DesyArchivePVManagerListener<V extends Serializable,
                                                   T extends ISystemVariable<V>> implements PVReaderListener {
// CHECKSTYLE ON: AbstractClassName

    private static final long _interval = 5*60*1000000000L;
    private static final Logger LOG = LoggerFactory.getLogger(DesyArchivePVManagerListener.class);
    private static final Logger EMAIL_LOG = LoggerFactory.getLogger("ErrorPerEmailLogger");
    private static final Logger CONN_LOG = LoggerFactory.getLogger("PvConnectionLogger");
    private static final Logger COUNT_LOG = LoggerFactory.getLogger("SampleCountLogger");

    public static Map<String, Integer> _samples = new HashMap<String, Integer>();

    private static long nanos = 0;
    private static Integer _countForAllPvs = 0;
    private Integer _countForPv = 0;
    private final ArchiveChannelBuffer<V, T> _buffer;
    private IServiceProvider _provider;
    private final String _channelName;
    private final ArchiveChannelId _channelId;
    private String _datatype;
    private volatile boolean _firstConnection;

    private String _startInfo;
    @SuppressWarnings("unused")
    private String _stopInfo;

    private final PVReader<?> _reader;
    /**
     * Constructor.
     */
    public DesyArchivePVManagerListener(@Nonnull final ArchiveChannelBuffer<V, T> buffer,
                                        @Nonnull final PVReader<?> reader,
                                        @Nonnull final IServiceProvider provider,
                                        @Nonnull final String name,
                                        @Nonnull final ArchiveChannelId id,
                                        @Nullable final String datatype) {
        _buffer = buffer;
        _reader = reader;
        _provider = provider;
        _channelName = name;
        _channelId = id;
        _datatype = datatype;
        _firstConnection = true;
        _samples.put(name, 0);
    }


    /**
     * {@inheritDoc}
     */
    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void pvChanged() {
        try {
            // TODO (bknerr) : ask Gabriele whether it is a good choice to separate the value
            // update callback from the calling/value providing instance
            final List<EpicsSystemVariable> sysVars = (List<EpicsSystemVariable>) _reader.getValue();
            if (_firstConnection && !sysVars.isEmpty()) {
                handleOnConnectionInformation(_provider, sysVars.get(0), _channelId, _buffer.isConnected(), _startInfo);
                _firstConnection = false;
            }
            if (nanos == 0) {
                nanos = sysVars.get(0).getTimestamp().getNanos() + _interval;
            }
            for (final EpicsSystemVariable sysVar : sysVars) {
                if (sysVar.getTimestamp().getNanos() > nanos) {
                    logPvCount();
                    COUNT_LOG.info("#Samples: " + _countForAllPvs + ", " + sysVar.getTimestamp().getNanos() + " > " + nanos);
                    _countForAllPvs = 0;
                    COUNT_LOG.info("##" + nanos + "##" + (nanos+_interval));
                    nanos = nanos + _interval;
                    resetPvCount();
                }
                _countForAllPvs++;
                _countForPv++;
                _samples.put(sysVar.getName(), _countForPv);
                handleValueUpdateInformation(sysVar);
            }
        } catch (final Throwable t) {
            LOG.error("Unexpected exception in PVListener for: {}:\n{}", _channelName, t.getMessage());
        }
    }

    private void logPvCount() {

        COUNT_LOG.info("-------Size: " + _samples.size());
        synchronized (_samples) {
            final Map<String, Integer> sampleCopy = new HashMap<String, Integer>();
            sampleCopy.putAll(_samples);
        }
        final Set<String> keySet = _samples.keySet();
        for (final String pvName : keySet) {
            COUNT_LOG.info(">>" + pvName + ">>" + _samples.get(pvName));
        }
    }


    private synchronized void resetPvCount() {
        final Set<String> keySet = _samples.keySet();
        for (final String pvName : keySet) {
            _samples.put(pvName, 0);
        }
    }


    @SuppressWarnings("rawtypes")
    private void handleOnConnectionInformation(@Nonnull final IServiceProvider provider,
                                               @Nonnull final EpicsSystemVariable pv,
                                               @Nonnull final ArchiveChannelId id,
                                               final boolean connected,
                                               @Nonnull final String info)
                                               throws EngineModelException, OsgiServiceUnavailableException, ArchiveServiceException {
        if (nanos == 0) {
            nanos = pv.getTimestamp().getNanos() + _interval;
            COUNT_LOG.info("##" + pv.getTimestamp().getNanos() + "##" + nanos);
        }
        _buffer.persistChannelStatusInfo(id, connected, info);

//        CONN_LOG.info(_channelName + ", " + connected);
        if(!validateAndPersistDatatype(provider, id, pv.getData())) {
            _buffer.stop("Datatype mismatch");
            return;
        }

        final EpicsMetaData metaData = pv.getMetaData();
        if (metaData != null) {
            handleMetaDataInfo(id, metaData);
        }
    }

    private boolean validateAndPersistDatatype(@Nonnull final IServiceProvider provider,
                                               @Nonnull final ArchiveChannelId id,
                                               @Nonnull final Object data) throws OsgiServiceUnavailableException,
                                                                                     ArchiveServiceException {

        final String actualTypeFromData = ArchiveTypeConversionSupport.createArchiveTypeStringFromData(data);

{'fraction_non_alphanumeric': 0.08010812139083426, 'fraction_numerical': 0.0031944956382848016, 'mean_word_length': 2.595979677490612, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5018539', 'n_tokens_mistral': 4365, 'n_tokens_neox': 4154, 'n_words': 1155}