text:
package org.apache.nifi.controller.repository;

import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.repository.claim.ContentClaim;
import org.apache.nifi.controller.repository.claim.ResourceClaim;
import org.apache.nifi.controller.repository.claim.ResourceClaimManager;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.rocksdb.RocksDBMetronome;
import org.apache.nifi.util.FormatUtils;
import org.apache.nifi.util.NiFiProperties;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wali.SerDe;
import org.wali.UpdateType;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.NumberFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * <p>
 * Implements FlowFile Repository using RocksDB as the backing store.
 * </p>
 */
public class RocksDBFlowFileRepository implements FlowFileRepository {

    private static final Logger logger = LoggerFactory.getLogger(RocksDBFlowFileRepository.class);

    private static final String FLOWFILE_PROPERTY_PREFIX = "nifi.flowfile.repository.";
    private static final String FLOWFILE_REPOSITORY_DIRECTORY_PREFIX = FLOWFILE_PROPERTY_PREFIX + "directory";

    private static final byte[] SWAP_LOCATION_SUFFIX_KEY = "swap.location.sufixes".getBytes(StandardCharsets.UTF_8);
    private static final byte[] SERIALIZATION_ENCODING_KEY = "serial.encoding".getBytes(StandardCharsets.UTF_8);
    private static final byte[] SERIALIZATION_HEADER_KEY = "serial.header".getBytes(StandardCharsets.UTF_8);
    static final byte[] REPOSITORY_VERSION_KEY = "repository.version".getBytes(StandardCharsets.UTF_8);
    static final byte[] VERSION_ONE_BYTES = "1.0".getBytes(StandardCharsets.UTF_8);
    private static final IllegalStateException NO_NEW_FLOWFILES = new IllegalStateException("Repository is not currently accepting new FlowFiles");
    private static final Runtime runtime = Runtime.getRuntime();
    private static final NumberFormat percentFormat = NumberFormat.getPercentInstance();

    private final Map<String, FlowFileQueue> queueMap = new HashMap<>();

    /**
     * Each property is defined by its name in the file and its default value
     */
    enum RocksDbProperty {
        //FlowFileRepo Configuration Parameters
        SYNC_WARNING_PERIOD("rocksdb.sync.warning.period", "30 seconds"),
        CLAIM_CLEANUP_PERIOD("rocksdb.claim.cleanup.period", "30 seconds"),
        DESERIALIZATION_THREADS("rocksdb.deserialization.threads", "16"),
        DESERIALIZATION_BUFFER_SIZE("rocksdb.deserialization.buffer.size", "1000"),
        SYNC_PERIOD("rocksdb.sync.period", "10 milliseconds"),
        ACCEPT_DATA_LOSS("rocksdb.accept.data.loss", "false"),
        ENABLE_STALL_STOP("rocksdb.enable.stall.stop", "false"),
        STALL_PERIOD("rocksdb.stall.period", "100 milliseconds"),
        STALL_FLOWFILE_COUNT("rocksdb.stall.flowfile.count", "800000"),
        STALL_HEAP_USAGE_PERCENT("rocksdb.stall.heap.usage.percent", "95%"),
        STOP_FLOWFILE_COUNT("rocksdb.stop.flowfile.count", "1100000"),
        STOP_HEAP_USAGE_PERCENT("rocksdb.stop.heap.usage.percent", "99.9%"),
        REMOVE_ORPHANED_FLOWFILES("rocksdb.remove.orphaned.flowfiles.on.startup", "false"),
        ENABLE_RECOVERY_MODE("rocksdb.enable.recovery.mode", "false"),
        RECOVERY_MODE_FLOWFILE_LIMIT("rocksdb.recovery.mode.flowfile.count", "5000"),

        //RocksDB Configuration Parameters
        DB_PARALLEL_THREADS("rocksdb.parallel.threads", "8"),
        MAX_WRITE_BUFFER_NUMBER("rocksdb.max.write.buffer.number", "4"),
        WRITE_BUFFER_SIZE("rocksdb.write.buffer.size", "256 MB"),
        LEVEL_O_SLOWDOWN_WRITES_TRIGGER("rocksdb.level.0.slowdown.writes.trigger", "20"),
        LEVEL_O_STOP_WRITES_TRIGGER("rocksdb.level.0.stop.writes.trigger", "40"),
        DELAYED_WRITE_RATE("rocksdb.delayed.write.bytes.per.second", "16 MB"),
        MAX_BACKGROUND_FLUSHES("rocksdb.max.background.flushes", "1"),
        MAX_BACKGROUND_COMPACTIONS("rocksdb.max.background.compactions", "1"),
        MIN_WRITE_BUFFER_NUMBER_TO_MERGE("rocksdb.min.write.buffer.number.to.merge", "1"),
        STAT_DUMP_PERIOD("rocksdb.stat.dump.period", "600 sec"),
        ;

        final String propertyName;
        final String defaultValue;

        RocksDbProperty(String propertyName, String defaultValue) {
            this.propertyName = FLOWFILE_PROPERTY_PREFIX + propertyName;
            this.defaultValue = defaultValue;
        }

        /**
         * @param niFiProperties The Properties file
         * @param timeUnit       The desired time unit
         * @return The property Value in the desired units
         */
        long getTimeValue(NiFiProperties niFiProperties, TimeUnit timeUnit) {
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue);
            long timeValue = 0L;
            try {
                timeValue = Math.round(FormatUtils.getPreciseTimeDuration(propertyValue, timeUnit));
            } catch (IllegalArgumentException e) {
                this.generateIllegalArgumentException(propertyValue, e);
            }
            return timeValue;
        }

        /**
         * @param niFiProperties The Properties file
         * @return The property value as a boolean
         */
        boolean getBooleanValue(NiFiProperties niFiProperties) {
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue);
            return Boolean.parseBoolean(propertyValue);
        }

        /**
         * @param niFiProperties The Properties file
         * @return The property value as an int
         */
        int getIntValue(NiFiProperties niFiProperties) {
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue);
            int returnValue = 0;
            try {
                returnValue = Integer.parseInt(propertyValue);
            } catch (NumberFormatException e) {
                this.generateIllegalArgumentException(propertyValue, e);
            }
            return returnValue;
        }

        /**
         * @param niFiProperties The Properties file
         * @return The property value as a number of bytes
         */
        long getByteCountValue(NiFiProperties niFiProperties) {
            long returnValue = 0L;
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue);
            try {
                double writeBufferDouble = DataUnit.parseDataSize(propertyValue, DataUnit.B);
                returnValue = (long) (writeBufferDouble < Long.MAX_VALUE ? writeBufferDouble : Long.MAX_VALUE);
            } catch (IllegalArgumentException e) {
                this.generateIllegalArgumentException(propertyValue, e);
            }
            return returnValue;
        }

        /**
         * @param niFiProperties The Properties file
         * @return The property value as a percent
         */
        double getPercentValue(NiFiProperties niFiProperties) {
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue).replace('%', ' ');
            double returnValue = 0.0D;
            try {
                returnValue = Double.parseDouble(propertyValue) / 100D;
                if (returnValue > 1.0D) {
                    this.generateIllegalArgumentException(propertyValue, null);
                }
            } catch (NumberFormatException e) {
                this.generateIllegalArgumentException(propertyValue, e);
            }
            return returnValue;
        }

        /**
         * @param niFiProperties The Properties file
         * @return The property value as a long
         */
        long getLongValue(NiFiProperties niFiProperties) {
            String propertyValue = niFiProperties.getProperty(this.propertyName, this.defaultValue);
            long returnValue = 0L;
            try {
                returnValue = Long.parseLong(propertyValue);
            } catch (NumberFormatException e) {
                this.generateIllegalArgumentException(propertyValue, e);
            }
            return returnValue;
        }

        void generateIllegalArgumentException(String badValue, Throwable t) {
            throw new IllegalArgumentException("The NiFi Property: [" + this.propertyName + "] with value: [" + badValue + "] is not valid", t);
        }

    }

    private final AtomicLong flowFileSequenceGenerator = new AtomicLong(0L);
    private final int deserializationThreads;
    private final int deserializationBufferSize;
    private final long claimCleanupMillis;
    private final ScheduledExecutorService housekeepingExecutor;
    private final AtomicReference<Collection<ResourceClaim>> claimsAwaitingDestruction = new AtomicReference<>(new ArrayList<>());
    private final RocksDBMetronome db;
    private ResourceClaimManager claimManager;
    private RepositoryRecordSerdeFactory serdeFactory;
    private SerDe<SerializedRepositoryRecord> serializer;
    private String serializationEncodingName;
    private byte[] serializationHeader;

    private final boolean acceptDataLoss;
    private final boolean enableStallStop;
    private final boolean removeOrphanedFlowFiles;
    private final boolean enableRecoveryMode;
    private final long recoveryModeFlowFileLimit;
    private final AtomicReference<SerDe<SerializedRepositoryRecord>> recordDeserializer = new AtomicReference<>();
    private final List<byte[]> recordsToRestore = Collections.synchronizedList(new LinkedList<>());

    private final ReentrantLock stallStopLock = new ReentrantLock();
    private final AtomicLong inMemoryFlowFiles = new AtomicLong(0L);
    volatile boolean stallNewFlowFiles = false;
    volatile boolean stopNewFlowFiles = false;
    private final long stallMillis;
    private final long stallCount;
    private final long stopCount;
    private final double stallPercentage;
    private final double stopPercentage;

    private final Set<String> swapLocationSuffixes = new HashSet<>(); // guarded by synchronizing on object itself

    /**
     * default no args constructor for service loading only.
     */
    public RocksDBFlowFileRepository() {
        deserializationThreads = 0;
        deserializationBufferSize = 0;
        claimCleanupMillis = 0;
        housekeepingExecutor = null;
        db = null;
        acceptDataLoss = false;
        enableStallStop = false;
        removeOrphanedFlowFiles = false;
        stallMillis = 0;
        stallCount = 0;
        stopCount = 0;
        stallPercentage = 0;
        stopPercentage = 0;
        enableRecoveryMode = false;
        recoveryModeFlowFileLimit = 0;
    }

    public RocksDBFlowFileRepository(final NiFiProperties niFiProperties) {
        deserializationThreads = RocksDbProperty.DESERIALIZATION_THREADS.getIntValue(niFiProperties);
        deserializationBufferSize = RocksDbProperty.DESERIALIZATION_BUFFER_SIZE.getIntValue(niFiProperties);

        claimCleanupMillis = RocksDbProperty.CLAIM_CLEANUP_PERIOD.getTimeValue(niFiProperties, TimeUnit.MILLISECONDS);
        housekeepingExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = Executors.defaultThreadFactory().newThread(r);
            thread.setDaemon(true);
            return thread;
        });

        acceptDataLoss = RocksDbProperty.ACCEPT_DATA_LOSS.getBooleanValue(niFiProperties);
        enableStallStop = RocksDbProperty.ENABLE_STALL_STOP.getBooleanValue(niFiProperties);

        removeOrphanedFlowFiles = RocksDbProperty.REMOVE_ORPHANED_FLOWFILES.getBooleanValue(niFiProperties);
        if (removeOrphanedFlowFiles) {
            logger.warn("The property \"{}\" is currently set to \"true\".  " +
                            "This can potentially lead to data loss, and should only be set if you are absolutely certain it is necessary.  " +
                            "Even then, it should be removed as soon as possible.",
                    RocksDbProperty.REMOVE_ORPHANED_FLOWFILES.propertyName);
        }
        stallMillis = RocksDbProperty.STALL_PERIOD.getTimeValue(niFiProperties, TimeUnit.MILLISECONDS);
        stallCount = RocksDbProperty.STALL_FLOWFILE_COUNT.getLongValue(niFiProperties);
        stopCount = RocksDbProperty.STOP_FLOWFILE_COUNT.getLongValue(niFiProperties);
        stallPercentage = RocksDbProperty.STALL_HEAP_USAGE_PERCENT.getPercentValue(niFiProperties);
        stopPercentage = RocksDbProperty.STOP_HEAP_USAGE_PERCENT.getPercentValue(niFiProperties);

        enableRecoveryMode = RocksDbProperty.ENABLE_RECOVERY_MODE.getBooleanValue(niFiProperties);
        recoveryModeFlowFileLimit = RocksDbProperty.RECOVERY_MODE_FLOWFILE_LIMIT.getLongValue(niFiProperties);
        if (enableRecoveryMode) {
            logger.warn("The property \"{}\" is currently set to \"true\" and  \"{}\" is set to  \"{}\".  " +
                            "This means that only {} FlowFiles will be loaded in to memory from the FlowFile repo at a time, " +
                            "allowing for recovery of a system encountering OutOfMemory errors (or similar).  " +
                            "This setting should be reset to \"false\" as soon as recovery is complete.",
                    RocksDbProperty.ENABLE_RECOVERY_MODE.propertyName, RocksDbProperty.RECOVERY_MODE_FLOWFILE_LIMIT.propertyName, recoveryModeFlowFileLimit, recoveryModeFlowFileLimit);
        }
        db = new RocksDBMetronome.Builder()
                .setStatDumpSeconds((int) (Math.min(RocksDbProperty.STAT_DUMP_PERIOD.getTimeValue(niFiProperties, TimeUnit.SECONDS), Integer.MAX_VALUE)))
                .setParallelThreads(RocksDbProperty.DB_PARALLEL_THREADS.getIntValue(niFiProperties))
                .setMaxWriteBufferNumber(RocksDbProperty.MAX_WRITE_BUFFER_NUMBER.getIntValue(niFiProperties))
                .setMinWriteBufferNumberToMerge(RocksDbProperty.MIN_WRITE_BUFFER_NUMBER_TO_MERGE.getIntValue(niFiProperties))
                .setWriteBufferSize(RocksDbProperty.WRITE_BUFFER_SIZE.getByteCountValue(niFiProperties))
                .setDelayedWriteRate(RocksDbProperty.DELAYED_WRITE_RATE.getByteCountValue(niFiProperties))
                .setLevel0SlowdownWritesTrigger(RocksDbProperty.LEVEL_O_SLOWDOWN_WRITES_TRIGGER.getIntValue(niFiProperties))
                .setLevel0StopWritesTrigger(RocksDbProperty.LEVEL_O_STOP_WRITES_TRIGGER.getIntValue(niFiProperties))
                .setMaxBackgroundFlushes(RocksDbProperty.MAX_BACKGROUND_FLUSHES.getIntValue(niFiProperties))
                .setMaxBackgroundCompactions(RocksDbProperty.MAX_BACKGROUND_COMPACTIONS.getIntValue(niFiProperties))
                .setSyncMillis(RocksDbProperty.SYNC_PERIOD.getTimeValue(niFiProperties, TimeUnit.MILLISECONDS))
                .setSyncWarningNanos(RocksDbProperty.SYNC_WARNING_PERIOD.getTimeValue(niFiProperties, TimeUnit.NANOSECONDS))
                .setStoragePath(getFlowFileRepoPath(niFiProperties))
                .setAdviseRandomOnOpen(false)
                .setCreateMissingColumnFamilies(true)
                .setCreateIfMissing(true)
                .setPeriodicSyncEnabled(!acceptDataLoss)
                .build();
    }

    /**
     * @param niFiProperties The Properties file
     * @return The path of the repo
     */
    static Path getFlowFileRepoPath(NiFiProperties niFiProperties) {
        for (final String propertyName : niFiProperties.getPropertyKeys()) {
            if (propertyName.startsWith(FLOWFILE_REPOSITORY_DIRECTORY_PREFIX)) {
                final String dirName = niFiProperties.getProperty(propertyName);
                return Paths.get(dirName);
            }
        }
        return null;
    }

    /**
     * @return The name of the File Store
     */
    @Override
    public String getFileStoreName() {
        try {
            return Files.getFileStore(db.getStoragePath()).name();
        } catch (IOException e) {
            return null;
        }
    }

    @Override
    public void initialize(final ResourceClaimManager claimManager) throws IOException {
        this.db.initialize();
        this.claimManager = claimManager;
        this.serdeFactory = new StandardRepositoryRecordSerdeFactory(claimManager);

        try {
            byte[] versionBytes = db.getConfiguration(REPOSITORY_VERSION_KEY);
            if (versionBytes == null) {
                db.putConfiguration(REPOSITORY_VERSION_KEY, VERSION_ONE_BYTES);
            } else if (!Arrays.equals(versionBytes, VERSION_ONE_BYTES)) {
                throw new IllegalStateException("Unknown repository version: " + new String(versionBytes, StandardCharsets.UTF_8));
            }

            byte[] serializationEncodingBytes = db.getConfiguration(SERIALIZATION_ENCODING_KEY);

            if (serializationEncodingBytes == null) {
                serializer = serdeFactory.createSerDe(null);
                serializationEncodingName = serializer.getClass().getName();
                db.putConfiguration(SERIALIZATION_ENCODING_KEY, serializationEncodingName.getBytes(StandardCharsets.UTF_8));
            } else {
                serializationEncodingName = new String(serializationEncodingBytes, StandardCharsets.UTF_8);
                serializer = serdeFactory.createSerDe(serializationEncodingName);
            }

            serializationHeader = db.getConfiguration(SERIALIZATION_HEADER_KEY);

            if (serializationHeader == null) {
                try (
                        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                        DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream)
                ) {
                    serializer.writeHeader(dataOutputStream);
                    serializationHeader = byteArrayOutputStream.toByteArray();
                    db.putConfiguration(SERIALIZATION_HEADER_KEY, serializationHeader);
                }
            }

            byte[] swapLocationSuffixBytes = db.getConfiguration(SWAP_LOCATION_SUFFIX_KEY);
            if (swapLocationSuffixBytes != null) {
                try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(swapLocationSuffixBytes);
                     ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream)) {
                    Object o = objectInputStream.readObject();
                    if (o instanceof Collection) {
                        ((Collection<?>) o).forEach(obj -> swapLocationSuffixes.add(obj.toString()));
                    }
                }
            }
        } catch (RocksDBException | ClassNotFoundException e) {
            throw new IOException(e);
        }

        housekeepingExecutor.scheduleWithFixedDelay(this::doHousekeeping, 0, claimCleanupMillis, TimeUnit.MILLISECONDS);

        logger.info("Initialized FlowFile Repository at {}", db.getStoragePath());
    }

    @Override
    public void close() throws IOException {
        if (housekeepingExecutor != null) {
            housekeepingExecutor.shutdownNow();
        }

        if (db != null) {
            db.close();
        }
    }

    /**
     * This method is scheduled by the housekeepingExecutor at the specified interval.
     * Catches Throwable so as not to suppress future executions of the ScheduledExecutorService
     */
    private void doHousekeeping() {
        try {
            doClaimCleanup();
            updateStallStop();
            doRecovery();
        } catch (Throwable t) {
            // catching Throwable so as not to suppress subsequent executions
            logger.error("Encountered problem during housekeeping", t);
        }
    }

    /**
     * Marks as destructible any claims that were held by records which have now been deleted.
     */
    private void doClaimCleanup() {
        Collection<ResourceClaim> claimsToDestroy;
        synchronized (claimsAwaitingDestruction) {
            if (Thread.currentThread().isInterrupted()) {
                return;
            }
            claimsToDestroy = claimsAwaitingDestruction.getAndSet(new ArrayList<>());
        }
        if (claimsToDestroy != null) {
            Collection<ResourceClaim> uniqueClaimsToDestroy = new HashSet<>(claimsToDestroy);

            try {
                if (!acceptDataLoss) {
                    db.waitForSync();
                } else {
                    db.forceSync();
                }
            } catch (InterruptedException | RocksDBException e) {
                synchronized (claimsAwaitingDestruction) {
                    // if there was an exception, put back the claims we were attempting to destroy
                    claimsAwaitingDestruction.get().addAll(uniqueClaimsToDestroy);
                    return;
                }
            }
            for (final ResourceClaim claim : uniqueClaimsToDestroy) {
                claimManager.markDestructable(claim);
            }
        }
    }

    /**
     * Updates the stalled and stopped status of the repository
     */
    void updateStallStop() {
        // if stall.stop logic is not enabled, return
        if (!enableStallStop) return;

        if (stallStopLock.tryLock()) {
            try {
                final long inMemoryFlowFiles = getInMemoryFlowFiles();

                if (inMemoryFlowFiles >= stopCount) {
                    stopNewFlowFiles = true;
                    stallNewFlowFiles = true;
                    logger.warn("Halting new FlowFiles because maximum FlowFile count ({}) has been exceeded.  Current count: {}",
                            new Object[]{stopCount, inMemoryFlowFiles});
                    return;
                }

                // calculate usage percentage
                final double freeMemory = runtime.freeMemory();
                final double maxMemory = runtime.maxMemory();
                final double usedPercentage = 1.0d - (freeMemory / maxMemory);

                if (usedPercentage >= stopPercentage) {
                    stopNewFlowFiles = true;
                    stallNewFlowFiles = true;
                    logger.warn("Halting new FlowFiles because maximum heap usage percentage ({}) has been exceeded.  Current usage: {}",
                            new Object[]{percentFormat.format(stopPercentage), percentFormat.format(usedPercentage)});
                    return;
                }

                if (inMemoryFlowFiles >= stallCount) {
                    stopNewFlowFiles = false;
                    stallNewFlowFiles = true;
                    logger.warn("Stalling new FlowFiles because FlowFile count stall threshold ({}) has been exceeded.  Current count: {}",
                            new Object[]{stallCount, inMemoryFlowFiles});
                    return;
                }

                if (usedPercentage >= stallPercentage) {
                    stopNewFlowFiles = false;
                    stallNewFlowFiles = true;
                    logger.warn("Stalling new FlowFiles because heap usage percentage threshold ({}) has been exceeded.  Current count: {}",
                            new Object[]{percentFormat.format(stallPercentage), percentFormat.format(usedPercentage)});
                    return;
                }

                if (stopNewFlowFiles || stallNewFlowFiles) {
                    logger.info("Resuming acceptance of new FlowFiles");
                    stopNewFlowFiles = false;
                    stallNewFlowFiles = false;
                }
            } finally {
                stallStopLock.unlock();
            }
        }
    }

    /**
     * If in recovery mode, restore more FlowFile if under the configured limit
     */
    synchronized void doRecovery() {

        // if we are not in recovery mode, return
        if (!enableRecoveryMode) return;

        SerDe<SerializedRepositoryRecord> deserializer = recordDeserializer.get();
        if (deserializer == null) {
            return; // initial load hasn't completed
        }

        if (recordsToRestore.isEmpty()) {
            logger.warn("Recovery has been completed.  " +
                            "The property \"{}\" is currently set to \"true\", but should be reset to \"false\" as soon as possible.",
                    RocksDbProperty.ENABLE_RECOVERY_MODE.propertyName);
            return;
        }

        logger.warn("The property \"{}\" is currently set to \"true\" and \"{}\" is set to \"{}\".  " +
                        "This means that only {} FlowFiles will be loaded into memory from the FlowFile repo at a time, " +
                        "allowing for recovery of a system encountering OutOfMemory errors (or similar).  " +
                        "This setting should be reset to \"false\" as soon as recovery is complete.  " +
                        "There are {} records remaining to be recovered.",
                RocksDbProperty.ENABLE_RECOVERY_MODE.propertyName, RocksDbProperty.RECOVERY_MODE_FLOWFILE_LIMIT.propertyName,
                recoveryModeFlowFileLimit, recoveryModeFlowFileLimit, getRecordsToRestoreCount());

        while (!recordsToRestore.isEmpty() && inMemoryFlowFiles.get() < recoveryModeFlowFileLimit) {
            try {
                byte[] key = recordsToRestore.get(0);
                byte[] recordBytes = db.get(key);
                if (recordBytes != null) {
                    try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(recordBytes);
                         DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream)) {
                        SerializedRepositoryRecord record = deserializer.deserializeRecord(dataInputStream, deserializer.getVersion());
                        final FlowFileRecord flowFile = record.getFlowFileRecord();

                        final FlowFileQueue queue = queueMap.get(record.getQueueIdentifier());
                        if (queue != null) {
                            queue.put(flowFile);
                            inMemoryFlowFiles.incrementAndGet();
                        }
                    }
                }
                recordsToRestore.remove(0);
            } catch (IOException | RocksDBException e) {
                logger.warn("Encountered exception during recovery", e);
            }
        }
    }

    long getInMemoryFlowFiles() {
        return inMemoryFlowFiles.get();
    }

    long getRecordsToRestoreCount() {
        return recordsToRestore.size();
    }

    /**
     * Updates the FlowFile repository with the given SerializedRepositoryRecords
     *
     * @param records the records to update the repository with
     * @throws IOException if update fails or a required sync is interrupted
     */
    @Override
    public void updateRepository(final Collection<RepositoryRecord> records) throws IOException {

        // verify records are valid
        int netIncrease = countAndValidateRecords(records);

        final boolean causeIncrease = netIncrease > 0;
        if (causeIncrease && stopNewFlowFiles) {
            updateStallStop();
            throw NO_NEW_FLOWFILES;
        }

        //update the db with the new records
        int syncCounterValue = updateRocksDB(records);
        inMemoryFlowFiles.addAndGet(netIncrease);

        try {
            // if we created data, but are at a threshold, delay and allow other data to try to get through
            if (causeIncrease && (stallNewFlowFiles || stopNewFlowFiles)) {
                Thread.sleep(stallMillis);
                updateStallStop();
            }

            // if we got a record indicating data creation, wait for it to be synced to disk before proceeding
            if (!acceptDataLoss && syncCounterValue > 0) {
                db.waitForSync(syncCounterValue);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException(e);
        }

        // determine any content claims that can be destroyed
        determineDestructibleClaims(records);
    }

    /**
     * Check that each record has the required elements
     *
     * @param records to be validated
     * @return the change in in-memory FlowFile count represented by this Collection
     */
    private int countAndValidateRecords(Collection<RepositoryRecord> records) {
        int inMemoryDelta = 0;
        for (RepositoryRecord record : records) {
            validateRecord(record);
            if (record.getType() == RepositoryRecordType.CREATE || record.getType() == RepositoryRecordType.SWAP_IN) {
                inMemoryDelta++;
            } else if (record.getType() == RepositoryRecordType.DELETE || record.getType() == RepositoryRecordType.SWAP_OUT) {
                inMemoryDelta--;
            }
        }
        return inMemoryDelta;
    }

    /**
     * Check that a record has the required elements
     *
     * @param record to be validated
     */
    private void validateRecord(RepositoryRecord record) {
        if (record.getType() != RepositoryRecordType.DELETE
                && record.getType() != RepositoryRecordType.CONTENTMISSING
                && record.getType() != RepositoryRecordType.CLEANUP_TRANSIENT_CLAIMS
                && record.getType() != RepositoryRecordType.SWAP_OUT
                && record.getDestination() == null) {
            throw new IllegalArgumentException("Record " + record + " has no destination and Type is " + record.getType());
        }
    }

    /**
     * Update the records in the RocksDB database
     *
     * @param records to be persisted
     * @return the value of the sync counter immediately after the last update which requires a sync
     */
    private int updateRocksDB(Collection<RepositoryRecord> records) throws IOException {

        // Partition records by UpdateType.
        // We do this because we want to ensure that records creating data are persisted first.
        // Additionally, remove records of type 'CLEANUP_TRANSIENT_CLAIMS' so they aren't sent to the db

        final Map<UpdateType, List<RepositoryRecord>> partitionedRecords = new HashMap<>();
        for (RepositoryRecord repositoryRecord : records) {
            if (repositoryRecord.getType() == RepositoryRecordType.CLEANUP_TRANSIENT_CLAIMS) {
                continue;
            }
            final UpdateType updateType = serdeFactory.getUpdateType(new LiveSerializedRepositoryRecord(repositoryRecord));
            partitionedRecords.computeIfAbsent(updateType, ut -> new ArrayList<>()).add(repositoryRecord);
        }

        int counterValue;

        try {
            // handle CREATE records
            putAll(partitionedRecords.get(UpdateType.CREATE));

            // handle SWAP_OUT records
            List<RepositoryRecord> swapOutRecords = partitionedRecords.get(UpdateType.SWAP_OUT);
            if (swapOutRecords != null) {
                for (final RepositoryRecord record : swapOutRecords) {
                    final SerializedRepositoryRecord serializedRecord = new LiveSerializedRepositoryRecord(record);
                    final String newLocation = serdeFactory.getLocation(serializedRecord);
                    final Long recordIdentifier = serdeFactory.getRecordIdentifier(serializedRecord);

                    if (newLocation == null) {
                        logger.error("Received Record (ID=" + recordIdentifier + ") with UpdateType of SWAP_OUT but " +
                                "no indicator of where the Record is to be Swapped Out to; these records may be " +
                                "lost when the repository is restored!");
                    } else {
                        delete(recordIdentifier);
                    }
                }
            }

            // handle SWAP_IN records
            List<RepositoryRecord> swapInRecords = partitionedRecords.get(UpdateType.SWAP_IN);
            if (swapInRecords != null) {
                for (final RepositoryRecord record : swapInRecords) {
                    final SerializedRepositoryRecord serialized = new LiveSerializedRepositoryRecord(record);

                    final String newLocation = serdeFactory.getLocation(serialized);
                    if (newLocation == null) {
                        final Long recordIdentifier = serdeFactory.getRecordIdentifier(serialized);
                        logger.error("Received Record (ID=" + recordIdentifier + ") with UpdateType of SWAP_IN but " +
                                "no indicator of where the Record is to be Swapped In from; these records may be " +
                                "duplicated when the repository is restored!");
                    }
                    put(record);
                }
            }

            // if a sync is required, get the current value of the sync counter
            counterValue = syncRequired(partitionedRecords) ? db.getSyncCounterValue() : -1;

            // handle UPDATE records
            putAll(partitionedRecords.get(UpdateType.UPDATE));

            // handle DELETE records
            deleteAll(partitionedRecords.get(UpdateType.DELETE));

        } catch (RocksDBException e) {
            throw new IOException(e);
        }

        return counterValue;
    }

    private boolean syncRequired(Map<UpdateType, List<RepositoryRecord>> recordMap) {
        for (UpdateType updateType : recordMap.keySet()) {
            if (updateType == UpdateType.CREATE || updateType == UpdateType.SWAP_OUT || updateType == UpdateType.SWAP_IN) {
                return true;
            }
        }
        return false;
    }

    private void deleteAll(List<RepositoryRecord> repositoryRecords) throws RocksDBException {
        if (repositoryRecords != null) {
            for (final RepositoryRecord record : repositoryRecords) {
                final SerializedRepositoryRecord serialized = new LiveSerializedRepositoryRecord(record);
                final Long id = serdeFactory.getRecordIdentifier(serialized);
                delete(id);
            }
        }
    }

    private void delete(Long recordId) throws RocksDBException {
        byte[] key = RocksDBMetronome.getBytes(recordId);
        db.delete(key);
    }

    private void putAll(List<RepositoryRecord> repositoryRecords) throws IOException, RocksDBException {
        if (repositoryRecords != null) {
            for (final RepositoryRecord record : repositoryRecords) {
                put(record);
            }
        }
    }

    private void put(RepositoryRecord record) throws IOException, RocksDBException {
        final Long recordIdentifier = serdeFactory.getRecordIdentifier(new LiveSerializedRepositoryRecord(record));
        byte[] key = RocksDBMetronome.getBytes(recordIdentifier);
        final byte[] serializedRecord = serialize(record);
        db.put(key, serializedRecord);
    }

    private byte[] serialize(RepositoryRecord record) throws IOException {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream)) {
            serializer.serializeRecord(new LiveSerializedRepositoryRecord(record), dataOutputStream);
            return byteArrayOutputStream.toByteArray();
        }
    }

    private void determineDestructibleClaims(Collection<RepositoryRecord> records) throws IOException {

        final Set<ResourceClaim> claimsToAdd = new HashSet<>();
        final Set<String> swapLocationsAdded = new HashSet<>();
        final Set<String> swapLocationsRemoved = new HashSet<>();
        for (final RepositoryRecord record : records) {
            updateClaimCounts(record);

            if (record.getType() == RepositoryRecordType.DELETE) {
                // For any DELETE record that we have, if claim is destructible, mark it so
                if (isDestructible(record.getCurrentClaim())) {
                    claimsToAdd.add(record.getCurrentClaim().getResourceClaim());
                }

                // If the original claim is different than the current claim and the original claim is destructible, mark it so
                if (shouldDestroyOriginal(record)) {
                    claimsToAdd.add(record.getOriginalClaim().getResourceClaim());
                }
            } else if (record.getType() == RepositoryRecordType.UPDATE) {
                // if we have an update, and the original is no longer needed, mark original as destructible
                if (shouldDestroyOriginal(record)) {
                    claimsToAdd.add(record.getOriginalClaim().getResourceClaim());
                }
            } else if (record.getType() == RepositoryRecordType.SWAP_OUT) {
                final String swapLocation = record.getSwapLocation();
                final String normalizedSwapLocation = normalizeSwapLocation(swapLocation);
                swapLocationsAdded.add(normalizedSwapLocation);
                swapLocationsRemoved.remove(normalizedSwapLocation);
            } else if (record.getType() == RepositoryRecordType.SWAP_IN) {
                final String swapLocation = record.getSwapLocation();
                final String normalizedSwapLocation = normalizeSwapLocation(swapLocation);
                swapLocationsRemoved.add(normalizedSwapLocation);
                swapLocationsAdded.remove(normalizedSwapLocation);
            }

            final List<ContentClaim> transientClaims = record.getTransientClaims();
            if (transientClaims != null) {
                for (final ContentClaim transientClaim : transientClaims) {
                    if (isDestructible(transientClaim)) {
                        claimsToAdd.add(transientClaim.getResourceClaim());
                    }
                }
            }
        }

        // If we have swapped files in or out, we need to ensure that we update our swapLocationSuffixes.
        if (!swapLocationsAdded.isEmpty() || !swapLocationsRemoved.isEmpty()) {
            synchronized (swapLocationSuffixes) {
                removeNormalizedSwapLocations(swapLocationsRemoved);
                addNormalizedSwapLocations(swapLocationsAdded);
            }
        }

        // add any new claims that can be destroyed to the list
        if (!claimsToAdd.isEmpty()) {
            synchronized (claimsAwaitingDestruction) {
                claimsAwaitingDestruction.get().addAll(claimsToAdd);
            }
        }
    }

    private void updateClaimCounts(final RepositoryRecord record) {
        final ContentClaim currentClaim = record.getCurrentClaim();
        final ContentClaim originalClaim = record.getOriginalClaim();
        final boolean claimChanged = !Objects.equals(currentClaim, originalClaim);

        if (record.getType() == RepositoryRecordType.DELETE || record.getType() == RepositoryRecordType.CONTENTMISSING) {
            decrementClaimCount(currentClaim);
        }

        if (claimChanged) {
            // records which have been updated - remove original if exists
            decrementClaimCount(originalClaim);
        }
    }

    private void decrementClaimCount(final ContentClaim claim) {
        if (claim == null) {
            return;
        }

        claimManager.decrementClaimantCount(claim.getResourceClaim());
    }

    /**
     * @param claim to be evaluated
     * @return true if the claim can be destroyed
     */
    private boolean isDestructible(final ContentClaim claim) {
        if (claim == null) {
            return false;
        }

        final ResourceClaim resourceClaim = claim.getResourceClaim();
        if (resourceClaim == null) {
            return false;
        }

        return !resourceClaim.isInUse();
    }

    /**
     * @param record to be evaluated
     * @return true if the original claim can be destroyed
     */
    private boolean shouldDestroyOriginal(RepositoryRecord record) {
        final ContentClaim originalClaim = record.getOriginalClaim();
        return isDestructible(originalClaim) && !originalClaim.equals(record.getCurrentClaim());
    }

    @Override
    public boolean isVolatile() {
        return false;
    }

    @Override
    public long getStorageCapacity() throws IOException {
        return db.getStorageCapacity();
    }

    @Override
    public long getUsableStorageSpace() throws IOException {
        return db.getUsableStorageSpace();
    }

    @Override
    public boolean isValidSwapLocationSuffix(final String swapLocationSuffix) {
        String normalizedSwapLocation = normalizeSwapLocation(swapLocationSuffix);
        synchronized (swapLocationSuffixes) {
            return swapLocationSuffixes.contains(normalizedSwapLocation);
        }
    }

    static String normalizeSwapLocation(final String swapLocation) {
        if (swapLocation == null) {
            return null;
        }

        final String normalizedPath = swapLocation.replace("\\", "/");
        final String withoutTrailing = (normalizedPath.endsWith("/") && normalizedPath.length() > 1) ? normalizedPath.substring(0, normalizedPath.length() - 1) : normalizedPath;
        final String pathRemoved = getLocationSuffix(withoutTrailing);

        return StringUtils.substringBefore(pathRemoved, ".");
    }

    private static String getLocationSuffix(final String swapLocation) {
        final int lastIndex = swapLocation.lastIndexOf("/");
        if (lastIndex < 0 || lastIndex >= swapLocation.length() - 1) {
            return swapLocation;
        }

        return swapLocation.substring(lastIndex + 1);
    }

    @Override
    public void swapFlowFilesOut(final List<FlowFileRecord> swappedOut, final FlowFileQueue queue,
                                 final String swapLocation) throws IOException {
        final List<RepositoryRecord> repoRecords = new ArrayList<>();
        if (swappedOut == null || swappedOut.isEmpty()) {
            return;
        }

        for (final FlowFileRecord swapRecord : swappedOut) {
            final RepositoryRecord repoRecord = new StandardRepositoryRecord(queue, swapRecord, swapLocation);
            repoRecords.add(repoRecord);
        }

        updateRepository(repoRecords);
        addRawSwapLocation(swapLocation);
        logger.info("Successfully swapped out {} FlowFiles from {} to Swap File {}", swappedOut.size(), queue, swapLocation);
    }

    @Override
    public void swapFlowFilesIn(final String swapLocation, final List<FlowFileRecord> swapRecords,
                                final FlowFileQueue queue) throws IOException {
        final List<RepositoryRecord> repoRecords = new ArrayList<>();

        for (final FlowFileRecord swapRecord : swapRecords) {
            final StandardRepositoryRecord repoRecord = new StandardRepositoryRecord(queue, swapRecord);
            repoRecord.setSwapLocation(swapLocation);   // set the swap file to indicate that it's being swapped in.
            repoRecord.setDestination(queue);

            repoRecords.add(repoRecord);
        }

        updateRepository(repoRecords);
        removeRawSwapLocation(swapLocation);
        logger.info("Repository updated to reflect that {} FlowFiles were swapped in to {}", new Object[]{swapRecords.size(), queue});
    }

    @Override
    public long loadFlowFiles(final QueueProvider queueProvider) throws IOException {

        final long startTime = System.nanoTime();

        queueMap.clear();
        for (final FlowFileQueue queue : queueProvider.getAllQueues()) {
            queueMap.put(queue.getIdentifier(), queue);
        }

        final ExecutorService recordDeserializationExecutor = Executors.newFixedThreadPool(deserializationThreads, r -> {
            Thread thread = Executors.defaultThreadFactory().newThread(r);
            thread.setDaemon(true);
            return thread;
        });

        // Create a queue which will hold the bytes of the records that have been read from disk and are awaiting deserialization
        final BlockingQueue<byte[]> recordBytesQueue = new ArrayBlockingQueue<>(deserializationBufferSize);

        final AtomicBoolean doneReading = new AtomicBoolean(false);
        final List<Future<Long>> futures = new ArrayList<>(deserializationThreads);

        RepositoryRecordSerdeFactory factory = new StandardRepositoryRecordSerdeFactory(claimManager);

        final AtomicInteger numFlowFilesMissingQueue = new AtomicInteger(0);
        final AtomicInteger recordCount = new AtomicInteger(0);
        final AtomicInteger recoveryModeRecordCount = new AtomicInteger(0);

        for (int i = 0; i < deserializationThreads; i++) {
            futures.add(recordDeserializationExecutor.submit(() -> {
                long localMaxId = 0;
                int localRecordCount = 0;
                final Set<String> localRecoveredSwapLocations = new HashSet<>();

                // Create deserializer in each thread
                final SerDe<SerializedRepositoryRecord> localDeserializer = factory.createSerDe(serializationEncodingName);
                try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(serializationHeader);
                     DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream)) {
                    localDeserializer.readHeader(dataInputStream);
                }

                while (!doneReading.get() || !recordBytesQueue.isEmpty()) {
                    byte[] value = recordBytesQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (value != null) {
                        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(value);
                             DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream)) {
                            SerializedRepositoryRecord record = localDeserializer.deserializeRecord(dataInputStream, localDeserializer.getVersion());

                            localRecordCount++;

                            // increment the count for the record
                            final ContentClaim claim = record.getContentClaim();
                            if (claim != null) {
                                claimManager.incrementClaimantCount(claim.getResourceClaim());
                            }

                            final long recordId = record.getFlowFileRecord().getId();
                            if (recordId > localMaxId) {
                                localMaxId = recordId;
                            }

                            if (record.getType().equals(RepositoryRecordType.SWAP_OUT)) {
                                localRecoveredSwapLocations.add(normalizeSwapLocation(record.getSwapLocation()));
                            }

                            final FlowFileRecord flowFile = record.getFlowFileRecord();
                            final FlowFileQueue queue = queueMap.get(record.getQueueIdentifier());
                            if (queue == null) {
                                if (!removeOrphanedFlowFiles) {
                                    throw new IOException("Found FlowFile in repository without a corresponding queue.  " +
                                            "This may indicate an issue syncing the flow.xml in a cluster.  " +
                                            "To resolve this issue you should restore the flow.xml.  " +
                                            "Alternatively, if removing data is acceptable, you can add the following to nifi.properties: \n\n" +
                                            "\t\t" + RocksDbProperty.REMOVE_ORPHANED_FLOWFILES.propertyName + "=true\n\n" +
                                            "...once this has allowed you to restart nifi, you should remove it from nifi.properties to prevent inadvertent future data loss.");
                                }

                                numFlowFilesMissingQueue.incrementAndGet();
                                try {
                                    final Long recordIdentifier = factory.getRecordIdentifier(record);
                                    byte[] key = RocksDBMetronome.getBytes(recordIdentifier);
                                    db.delete(key);
                                } catch (RocksDBException e) {
                                    logger.warn("Could not clean up repository", e);
                                }
                            } else {
                                // verify we're supposed to enqueue the FlowFile
                                if (!enableRecoveryMode) {
                                    queue.put(flowFile);
                                } else if (recoveryModeRecordCount.incrementAndGet() <= recoveryModeFlowFileLimit) {
                                    queue.put(flowFile);
                                } else {
                                    final Long recordIdentifier = factory.getRecordIdentifier(record);
                                    byte[] key = RocksDBMetronome.getBytes(recordIdentifier);
                                    recordsToRestore.add(key);
                                }
                            }
                        }
                    }
                }
                recordCount.addAndGet(localRecordCount);
                addNormalizedSwapLocations(localRecoveredSwapLocations);
                return localMaxId;

            }));
        }

        long maxId = 0;
        RocksIterator rocksIterator = db.getIterator();
        rocksIterator.seekToFirst();
        long counter = 0;
        long totalRecords = 0;
        try {
            while (rocksIterator.isValid()) {
                if (recordBytesQueue.offer(rocksIterator.value(), 10, TimeUnit.SECONDS)) {
                    rocksIterator.next();
                    if (++counter == 5_000) { // periodically report progress
                        totalRecords += counter;
                        counter = 0;
                        logger.info("Read {} records from disk", totalRecords);
                    }
                } else {
                    // couldn't add to the queue in a timely fashion... make sure there are no exceptions from the consumers
                    for (Future<Long> f : futures) {
                        // the only way it could be done at this point is through an exception
                        // (because we haven't yet set doneReading = true)
                        if (f.isDone()) {
                            f.get(); // this will throw the exception
                        }
                    }
                    logger.warn("Failed to add record bytes to queue.  Will keep trying...");
                }
            }

            doneReading.set(true);
            totalRecords += counter;
            logger.info("Finished reading from rocksDB.  Read {} records from disk", totalRecords);

            for (Future<Long> f : futures) {
                long futureMax = f.get(); // will wait for completion (or exception)
                if (futureMax > maxId) {
                    maxId = futureMax;
                }
            }

            logger.info("Finished deserializing {} records", recordCount.get());

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException(e);
        } catch (ExecutionException e) {
            throw new IOException(e);
        } finally {
            recordDeserializationExecutor.shutdownNow();
        }

        // Set the AtomicLong to 1 more than the max ID so that calls to #getNextFlowFileSequence() will
        // return the appropriate number.
        flowFileSequenceGenerator.set(maxId + 1);

        int flowFilesInQueues = recordCount.get() - numFlowFilesMissingQueue.get();
        inMemoryFlowFiles.set(!enableRecoveryMode ? flowFilesInQueues
                : Math.min(flowFilesInQueues, recoveryModeFlowFileLimit));
        logger.info("Successfully restored {} FlowFiles in {} milliseconds using {} threads",
                getInMemoryFlowFiles(),
                TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime),
                deserializationThreads);
        if (logger.isDebugEnabled()) {
            synchronized (this.swapLocationSuffixes) {
                logger.debug("Recovered {} Swap Files: {}", swapLocationSuffixes.size(), swapLocationSuffixes);
            }
        }
        if (numFlowFilesMissingQueue.get() > 0) {
            logger.warn("On recovery, found {} FlowFiles whose queue no longer exists.  These FlowFiles have been dropped.", numFlowFilesMissingQueue);
        }

        final SerDe<SerializedRepositoryRecord> deserializer = factory.createSerDe(serializationEncodingName);

        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(serializationHeader);
             DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream)) {
            deserializer.readHeader(dataInputStream);
        }

        if (enableRecoveryMode) {
            recordDeserializer.set(deserializer);
        }

        return maxId;
    }

    @Override
    public Set<String> findQueuesWithFlowFiles(final FlowFileSwapManager flowFileSwapManager) throws IOException {
        return null;
    }

    private void addRawSwapLocation(String rawSwapLocation) throws IOException {
        addRawSwapLocations(Collections.singleton(rawSwapLocation));
    }

    private void addRawSwapLocations(Collection<String> rawSwapLocations) throws IOException {
        addNormalizedSwapLocations(rawSwapLocations.stream().map(RocksDBFlowFileRepository::normalizeSwapLocation).collect(Collectors.toSet()));
    }

    private void addNormalizedSwapLocations(Collection<String> normalizedSwapLocations) throws IOException {
        synchronized (this.swapLocationSuffixes) {
            this.swapLocationSuffixes.addAll(normalizedSwapLocations);
            persistSwapLocationSuffixes();
        }
    }

    private void removeRawSwapLocation(String rawSwapLocation) throws IOException {
        removeRawSwapLocations(Collections.singleton(rawSwapLocation));
    }

    private void removeRawSwapLocations(Collection<String> rawSwapLocations) throws IOException {
        removeNormalizedSwapLocations(rawSwapLocations.stream().map(RocksDBFlowFileRepository::normalizeSwapLocation).collect(Collectors.toSet()));
    }

    private void removeNormalizedSwapLocations(Collection<String> normalizedSwapLocations) throws IOException {
        synchronized (this.swapLocationSuffixes) {
            this.swapLocationSuffixes.removeAll(normalizedSwapLocations);
            persistSwapLocationSuffixes();
        }
    }

    private void persistSwapLocationSuffixes() throws IOException {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream)
        ) {
            objectOutputStream.writeObject(swapLocationSuffixes);
            db.putConfiguration(SWAP_LOCATION_SUFFIX_KEY, byteArrayOutputStream.toByteArray());
        } catch (RocksDBException e) {
            throw new IOException(e);
        }
    }

    @Override
    public void updateMaxFlowFileIdentifier(final long maxId) {
        while (true) {
            final long currentId = flowFileSequenceGenerator.get();
            if (currentId >= maxId) {
                return;
            }

            final boolean updated = flowFileSequenceGenerator.compareAndSet(currentId, maxId);
            if (updated) {
                return;
            }
        }
    }

    @Override
    public long getNextFlowFileSequence() {
        return flowFileSequenceGenerator.getAndIncrement();
    }

    @Override
    public long getMaxFlowFileIdentifier() {
        // flowFileSequenceGenerator is 1 more than the MAX so that we can call #getAndIncrement on the AtomicLong
        return flowFileSequenceGenerator.get() - 1;
    }
}


text:
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML+RDFa 1.1//EN" "http://www.w3.org/MarkUp/DTD/xhtml-rdfa-2.dtd">

  <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
  <title>Timeline Layer - HuTime API Manual</title>
  <link href="/manual/common/manual.css" rel="stylesheet" type="text/css"/>
  <meta property="dcterms:title" content="HuTime Manual" />
  <meta property="dcterms:description" content="User's manual about usage of HuTime." />

HuTime API Manual

Timeline Layer

1. Overview

A timeline (e.g., chronological table) is displayed with a timeline layer (HuTime.TLineLayer). The timeline layer, a class derived from the Layer class, can basically be used like other layers, for example, when you put it on a panel. Source data for a timeline can be separately given in a different form, such as a CSV file. You can use a record set for the timeline (HuTime.TLineRecordset) to specify which items (columns) in the source data correspond to the start and end times, as well as which items are displayed.

2. Source Data for a Timeline

We recommend for easier handling that you use a file whose first line is the title row for the csv (comma separated value) file or tsv (tab separated value) file. Here is an example of source data, which contains the list of typhoons in 2015 (the first several rows only):

Number,From,To,Name,PeakWS,PeakAP,Category
1,2015-01-14,2015-01-19,Mekkhala,110,975,1
2,2015-02-08,2015-02-11,Higos,165,940,4
3,2015-03-11,2015-03-17,Bavi,85,990,0
4,2015-03-28,2015-04-05,Maysak,195,910,5
5,2015-04-04,2015-04-04,Haishen,65,998,0

3. Creating a Layer Object

You specify the URL of the source data for the chart when creating an object. The height, as well as the top and bottom margins, of the layer can be specified similarly to other layers.

Constructor

HuTime.TLineLayer([recordset [, vBreadth [, vMarginTop [, vMarginBottom]]]])

recordset

Type: HuTime.TLineRecordset

The record set that contains the source data. You can access the record set specified in the constructor as the first array element in the recordsets property (layer.recordsets[0]). If a record set is not specified, you need to add a record set separately using the appendRecordset method.

vBreadth

Type: Number

The height of the layer (in pixels).

vMarginTop

Type: Number

The size of the top margin of the layer (in pixels).

vMarginBottom

Type: Number

The size of the bottom margin of the layer (in pixels).

Overview of the Record Set

The record set for the timeline is used.

HuTime.CalendarTLineRecordset(source, tBegin, tEnd, label [, calendarId, [, rangeStyle [, labelStyle]]])

source

Type: String or HuTime.StreamReaderBase

When a string is entered, it is interpreted as the URL of the source data for the record set, a new StreamReader object is created, and then the data is read.

tBegin

Type: String or number

The item name or the column number of the start date and time in the record.

tEnd

Type: String or number

The item name or the column number of the end date and time in the record.

label

Type: String or number

The item name or the column number for the strings that are displayed in the timeline, in the record.

calendarId

Type: String

The calendar ID of the start and end dates and times. If omitted, it is set to the Gregorian calendar.

rangeStyle

Type: HuTime.FigureStyle

The format of the band that shows the record duration on the timeline. If omitted, it is formatted with the default for the timeline layer.

labelStyle

Type: HuTime.StringStyle

The format of the record label on the timeline. If omitted, it is formatted with the default for the timeline layer.

4. Adding a Layer to the Panel and Displaying

Just like other layers, you can add a timeline layer to a panel using the appendLayer method of a Panel object. The object is drawn according to the sequence of the redraw method, but if the source data has not been read completely, it takes time to show the object.

An example of displaying a timeline layer (see the sample code below). The source data is the list of typhoons in 2015, which is shown in "2. Source Data for a Timeline."

Sample code (click here for a running example):

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <script type="text/javascript" src="http://web.hutime.org/js/HuTime.min.js"></script>
  <title>Sample Code</title>
</head>
<body>
<div id="timeline"></div>
<script type="text/javascript">
    var ht = new HuTime();
    var pnls = new HuTime.PanelCollection(200, 400);
    ht.appendPanelCollection(pnls, document.getElementById("timeline"));
    var pnl = new HuTime.TilePanel(200);
    pnls.appendPanel(pnl);

    var lyr = new HuTime.TLineLayer(
        new HuTime.CalendarTLineRecordset("Typhoon2015.csv", "From", "To", "Name"));
    pnl.appendLayer(lyr);

    ht.redraw(HuTime.timeToJd(2015, 1, 1), HuTime.timeToJd(2016, 1, 1));
</script>
</body>
</html>

5. Formatting

In a timeline, you can format the band for duration, and labels. A timeline can be formatted on a record-set basis. When a format value is passed to the constructor of a record set as an argument, it will be the default format of that record set. The following figure shows an example with some format set to the sample code in "4. Adding a Layer to the Panel and Displaying."

An example of formatting the timeline layer (see the sample code below). In the example above, the timeline of the sample code in "4. Adding a Layer to the Panel and Displaying” is formatted by passing arguments to the constructor of HuTime.CalendarTLineRecordset.

Sample code (click here for a running example). The null value is specified for the calendarId argument of the HuTime.CalendarTLineRecordset's constructor. A calendar scale layer (variable: scl) is also added.

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <script type="text/javascript" src="http://web.hutime.org/js/HuTime.min.js"></script>
  <title>Sample Code</title>
</head>
<body>
<div id="timeline"></div>
<script type="text/javascript">
    var ht = new HuTime();
    var pnls = new HuTime.PanelCollection(200, 400);
    ht.appendPanelCollection(pnls, document.getElementById("timeline"));
    var pnl = new HuTime.TilePanel(200);
    pnls.appendPanel(pnl);

    var lyr = new HuTime.TLineLayer(
        new HuTime.CalendarTLineRecordset("Typhoon2015.csv", "From", "To", "Name", null,
            new HuTime.FigureStyle("lime"), new HuTime.StringStyle(20, "red")));
    pnl.appendLayer(lyr);

    var scl = new HuTime.CalendarScaleLayer(50, null, 0);
    pnl.appendLayer(scl);

    ht.redraw(HuTime.timeToJd(2015, 1, 1), HuTime.timeToJd(2016, 1, 1));
</script>
</body>
</html>

6. Displaying Multiple Data Sets

You can add multiple record sets to a timeline layer by using the appendRecordset property. When different record sets have different formats, a timeline can show the differences of record sets.

appendRecordset(recordset)

Adds the record set to a timeline layer.

recordset

Type: HuTime.TLineRecordset

The record set to be added.

An example of the timeline layer with multiple record sets added to it (see the sample code below). The Hurricane2015.csv file (the list of hurricanes in 2015) is added as new source data using the appendRecordset method, and the band for duration is formatted in light blue and the label in red.

Sample code (click here for a running example):

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <script type="text/javascript" src="http://web.hutime.org/js/HuTime.min.js"></script>
  <title>Sample Code (TLine Layer Style Setting)</title>
</head>
<body>
<div id="timeline"></div>
<script type="text/javascript">
    var ht = new HuTime();
    var pnls = new HuTime.PanelCollection(200, 400);
    ht.appendPanelCollection(pnls, document.getElementById("timeline"));
    var pnl = new HuTime.TilePanel(200);
    pnls.appendPanel(pnl);

    var lyr = new HuTime.TLineLayer(
        new HuTime.CalendarTLineRecordset("Typhoon2015.csv", "From", "To", "Name"));
    pnl.appendLayer(lyr);

    lyr.appendRecordset(
        new HuTime.CalendarTLineRecordset("Hurricane2015.csv", "From", "To", "Name", null,
            new HuTime.FigureStyle("cyan"), new HuTime.StringStyle(14, "red")));

    var scl = new HuTime.CalendarScaleLayer(50, null, 0);
    pnl.appendLayer(scl);

    ht.redraw(HuTime.timeToJd(2015, 1, 1), HuTime.timeToJd(2016, 1, 1));
</script>
</body>
</html>

7. Formatting Individual Records

To color-code data in a timeline based on a particular item in the same record set, set a color-coding function for the rangeStyle property of the record set object. A HuTime.TLineRecord object is passed to the set function as an argument, and return a HuTime.FigureStyle object based on its contents.

In the figure and sample code below, if the data item Category in the source data has a value of 4 or more, the record is displayed in yellow, and it has a value of 3 or less, the record is displayed in light blue. First, two formats (styleStrong and styleWeak) are defined. Next, a record set object is created additionally, and the settings for reading the data item Category are added to that record set. Finally, a processing for outputting the formats to the rangeStyle property of the created record set is described.

An example of formatting individual record (see the sample code below). Typhoons in category 4 or higher are displayed in yellow, and those in category 3 or lower in light blue.

Sample code (click here for a running example):

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <script type="text/javascript" src="http://web.hutime.org/js/HuTime.min.js"></script>
  <title>Sample Code</title>
</head>
<body>
<div id="timeline"></div>
<script type="text/javascript">
    var ht = new HuTime();
    var pnls = new HuTime.PanelCollection(200, 400);
    ht.appendPanelCollection(pnls, document.getElementById("timeline"));
    var pnl = new HuTime.TilePanel(200);
    pnls.appendPanel(pnl);

    var styleStrong = new HuTime.FigureStyle("yellow");
    var styleWeak = new HuTime.FigureStyle("cyan");
    var recordset = new HuTime.CalendarTLineRecordset("Typhoon2015.csv", "From", "To", "Name");
    recordset.recordSettings.appendDataSetting(new HuTime.RecordDataSetting("Category"));
    recordset.rangeStyle = function(record) {
        if (!record.data["Category"])
            return defaultRecordStyle;
        if (record.data["Category"].text >= 4)
            return styleStrong;
        else
            return styleWeak;
    };

    var lyr = new HuTime.TLineLayer(recordset);
    pnl.appendLayer(lyr);

    var scl = new HuTime.CalendarScaleLayer(50, null, 0);
    pnl.appendLayer(scl);

    ht.redraw(HuTime.timeToJd(2015, 1, 1), HuTime.timeToJd(2016, 1, 1));
</script>
</body>
</html>

8. Events Handling for Records

The actions listed below on records on a timeline can be supplemented as an event on a timeline layer. The clicked record is stored in records of the event object as an array. When multiple overlapped records are displayed, the corresponding records are all stored in records. An event handler is added using addEventListener, just like other events.

        Action
        Event Type

        Click
        plotclick

        Double-click
        plotdblclick

In the example below, the recordClick function is added to the timeline layer as plotclick event handling. The recordClick function concatenates the labels of clicked records and displays the combined label in the pop-up window.

Sample code (click here for a running example):

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <script type="text/javascript" src="http://web.hutime.org/js/HuTime.min.js"></script>
  <title>Sample Code (TLine Layer Style Setting)</title>
</head>
<body>
<div id="timeline"></div>
<script type="text/javascript">
    var ht = new HuTime();
    var pnls = new HuTime.PanelCollection(200, 400);
    ht.appendPanelCollection(pnls, document.getElementById("timeline"));
    var pnl = new HuTime.TilePanel(200);
    pnls.appendPanel(pnl);

    var lyr = new HuTime.TLineLayer(
        new HuTime.CalendarTLineRecordset("Typhoon2015.csv", "From", "To", "Name"));
    lyr.addEventListener("plotclick", recordClick);
    pnl.appendLayer(lyr);

    var scl = new HuTime.CalendarScaleLayer(50, null, 0);
    pnl.appendLayer(scl);

    ht.redraw(HuTime.timeToJd(2015, 1, 1), HuTime.timeToJd(2016, 1, 1));

    function recordClick(ev) {
        var text = "";
        for (var i = 0; i < ev.records.length; ++i) {
            text += ev.records[i].record.data["Name"].text + " ";
        }
        alert(text);
    }
</script>
</body>
</html>

redpajama_set_name: RedPajamaGithub


text:
using System;
using System.Collections.Generic;
using System.Linq;
using BlueSheep.Common.IO;

namespace BlueSheep.Common.Protocol.Types
{

public class BasicAllianceInformations : AbstractSocialGroupInfos
{

public new const short ID = 419;
public override short TypeId
{
    get { return ID; }
}

public int allianceId;
        public string allianceTag;

public BasicAllianceInformations()
{
}

public BasicAllianceInformations(int allianceId, string allianceTag)
        {
            this.allianceId = allianceId;
            this.allianceTag = allianceTag;
        }

public override void Serialize(BigEndianWriter writer)
{

base.Serialize(writer);
            writer.WriteVarInt(allianceId);
            writer.WriteUTF(allianceTag);

}

public override void Deserialize(BigEndianReader reader)
{

base.Deserialize(reader);
            allianceId = reader.ReadVarInt();
            if (allianceId < 0)
                throw new Exception("Forbidden value on allianceId = " + allianceId + ", it doesn't respect the following condition : allianceId < 0");
            allianceTag = reader.ReadUTF();

}

}

}


text:
#ifndef __BACKPORT_ASM_ERRNO_H
#define __BACKPORT_ASM_ERRNO_H
#include_next <asm/errno.h>

#ifndef ERFKILL
#if !defined(CONFIG_ALPHA) && !defined(CONFIG_MIPS) && !defined(CONFIG_PARISC) && !defined(CONFIG_SPARC)
#define ERFKILL		132	/* Operation not possible due to RF-kill */
#endif
#ifdef CONFIG_ALPHA
#define ERFKILL		138	/* Operation not possible due to RF-kill */
#endif
#ifdef CONFIG_MIPS
#define ERFKILL		167	/* Operation not possible due to RF-kill */
#endif
#ifdef CONFIG_PARISC
#define ERFKILL		256	/* Operation not possible due to RF-kill */
#endif
#ifdef CONFIG_SPARC
#define ERFKILL		134	/* Operation not possible due to RF-kill */
#endif
#endif

#endif /* __BACKPORT_ASM_ERRNO_H */


text:
// $Header$

#include <ga-mpi/GAStringGenome.h>

template <> const char *
GA1DArrayAlleleGenome<char>::className() const {return "GAStringGenome";}
template <> int
GA1DArrayAlleleGenome<char>::classID() const {return GAID::StringGenome;}

template <> GA1DArrayAlleleGenome<char>::
GA1DArrayAlleleGenome(unsigned int length, const GAAlleleSet<char> & s,
		      GAGenome::Evaluator f, void * u) :
GA1DArrayGenome<char>(length, f, u){
  naset = 1;
  aset = new GAAlleleSet<char>[1];
  aset[0] = s;

  initializer(DEFAULT_STRING_INITIALIZER);
  mutator(DEFAULT_STRING_MUTATOR);
  comparator(DEFAULT_STRING_COMPARATOR);
  crossover(DEFAULT_STRING_CROSSOVER);
}

template <> GA1DArrayAlleleGenome<char>::
GA1DArrayAlleleGenome(const GAAlleleSetArray<char> & sa,
		      GAGenome::Evaluator f, void * u) :
GA1DArrayGenome<char>(sa.size(), f, u){
  naset = sa.size();
  aset = new GAAlleleSet<char>[naset];
  for(int i=0; i<naset; i++)
    aset[i] = sa.set(i);

  initializer(DEFAULT_STRING_INITIALIZER);
  mutator(DEFAULT_STRING_MUTATOR);
  comparator(DEFAULT_STRING_COMPARATOR);
  crossover(DEFAULT_STRING_CROSSOVER);
}

template <>
GA1DArrayAlleleGenome<char>::~GA1DArrayAlleleGenome(){
  delete [] aset;
}

#ifdef GALIB_USE_STREAMS
// The read specialization takes in each character whether it is whitespace or
// not and stuffs it into the genome.  This is unlike the default array read.
template <> int
GA1DArrayAlleleGenome<char>::read(STD_ISTREAM & is)
{
  unsigned int i=0;
  char c;
  do{
    is.get(c);
    if(!is.fail()) gene(i++, c);
  } while(!is.fail() && !is.eof() && i < nx);

  if(is.eof() && i < nx){
    GAErr(GA_LOC, className(), "read", gaErrUnexpectedEOF);
    is.clear(STD_IOS_BADBIT | is.rdstate());
    return 1;
  }
  return 0;
}

// Unlike the base array genome, here when we write out we don't put any
// whitespace between genes.  No newline at end of it all.
template <> int
GA1DArrayAlleleGenome<char>::write(STD_OSTREAM & os) const
{
  for(unsigned int i=0; i<nx; i++)
    os << gene(i);
  return 0;
}
#endif

// force instantiations of this genome type.
//
// These must be included _after_ the specializations because some compilers
// get all wigged out about the declaration/specialization order.  Note that
// some compilers require a syntax different than others when forcing the
// instantiation (i.e. GNU wants the 'template class', borland does not).
#ifndef GALIB_USE_AUTO_INST
#include <ga-mpi/GAAllele.C>
#include <ga-mpi/GA1DArrayGenome.C>

#if defined(__BORLANDC__)
#define GALIB_STRINGGENOME_TEMPLATE_PREFACE
#else
#define GALIB_STRINGGENOME_TEMPLATE_PREFACE template class
#endif

GALIB_STRINGGENOME_TEMPLATE_PREFACE GAAlleleSet<char>;
GALIB_STRINGGENOME_TEMPLATE_PREFACE GAAlleleSetCore<char>;
GALIB_STRINGGENOME_TEMPLATE_PREFACE GAAlleleSetArray<char>;

GALIB_STRINGGENOME_TEMPLATE_PREFACE GAArray<char>;
GALIB_STRINGGENOME_TEMPLATE_PREFACE GA1DArrayGenome<char>;
GALIB_STRINGGENOME_TEMPLATE_PREFACE GA1DArrayAlleleGenome<char>;

#endif


text:
<?php namespace EC\Api;
defined('_ESPADA') or die(NO_ACCESS);

use E, EC;

class CResult_Base
{

    const SUCCESS       = 0;
    const FAILURE       = 1;
    const ERROR         = 2;

    static public function Success_Base($type, $message = '')
    {
        if ($type === 'json')
            return new CResult(self::SUCCESS, $message);
        else if ($type === 'bytes')
            return new CResult_Bytes(self::SUCCESS, $message);

        throw new \Exception('Unknown result type.');
    }

    static public function Failure_Base($type, $message = '')
    {
        if ($type === 'json')
            return new CResult(self::FAILURE, $message);
        else if ($type === 'bytes')
            return new CResult_Bytes(self::FAILURE, $message);

        throw new \Exception('Unknown result type.');
    }

    static public function Error_Base($type, $message = '')
    {
        if ($type === 'json')
            return new CResult(self::ERROR, $message);
        else if ($type === 'bytes')
            return new CResult_Bytes(self::ERROR, $message);

        throw new \Exception('Unknown result type.');
    }

    private $result = null;
    private $message = null;
    private $debug = null;

    public function __construct(int $result, string $message)
    {
        $this->result = $result;
        $this->message = $message;
    }

    public function debug($message)
    {
        if (EDEBUG) {
            if ($this->debug === null)
                $this->debug = [];

            $this->debug[] = $message;
        }

        return $this;
    }

    public function isSuccess()
    {
        return $this->result === self::SUCCESS;
    }

    public function isFailure()
    {
        return $this->result === self::FAILURE;
    }

    public function isError()
    {
        return $this->result === self::ERROR;
    }

    public function getDebug()
    {
        return $this->debug;
    }

    public function getMessage()
    {
        return $this->message;
    }

    public function getResult()
    {
        return $this->result;
    }

}


text:
define([
	"intern", "intern!object", "intern/dojo/node!leadfoot/helpers/pollUntil", "intern/chai!assert",
	"intern/dojo/node!leadfoot/keys", "require"
], function (intern, registerSuite, pollUntil, assert, keys, require) {

	function loadFile(remote, url) {
		return remote
			.get(require.toUrl(url))
			.then(pollUntil("return ready ? true : null;", [],
				intern.config.WAIT_TIMEOUT, intern.config.POLL_INTERVAL));
	}

	function getTransformX(v) {
		var m = v.match(/matrix\(.*,.*,.*,.*,\s*(.*)\s*,.*\)/);
		if (m && m.length > 0) {
			return m[1];
		} else {
			m = v.match(/matrix3d\(.*,.*,.*,.*,.*,.*,.*,.*,.*,.*,.*,.*,\s*(.*)\s*,.*,.*,.*\)/);
			if (m && m.length > 0) {
				return m[1];
			} else {
				return "unknown transform";
			}
		}
	}

	registerSuite({
		name: "SwapView - functional",

		"SwapView load": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			return loadFile(remote, "./SwapView.html").findById("sv").then(function () {
				pollUntil("return document.getElementById('sv').className == 'd-swap-view d-view-stack'", [], 2000,
					intern.config.POLL_INTERVAL);
			})
				.execute("return document.getElementById('sv').children.length;")
				.then(function (v) {
					assert.equal(v, 3, "Wrong number of children in SwapView");
				})
			;
		},

		"ViewIndicator initialization": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			return remote
				.execute("return document.getElementById('vi').className;")
				.then(function (v) {
					assert.equal(v, "d-view-indicator", "Wrong class for ViewIndicator");
				})
				.execute("return document.getElementById('vi').children.length;")
				.then(function (v) {
					assert.equal(v, 3, "Wrong number of children in ViewIndicator");
				})
				.execute("return document.getElementById('vi').children[0].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 0");
					assert.include(v, "-d-view-indicator-dot-selected", "Missing class for vi child 0");
				})
				.execute("return document.getElementById('vi').children[1].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 1");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 1");
				})
				.execute("return document.getElementById('vi').children[2].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 2");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 2");
				})
			;
		},

		"SwapView swipe gesture (right->left)": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
			    return this.skip("SafariDriver doesn't support moveTo.");
			} else {
				return remote
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 200, 100);
					})
					.end()
					.pressMouseButton()
					.sleep(50)
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 195, 100);
					})
					.end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.notInclude(v, " -d-swap-view-drag", "Swipe started before drag threshold");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 189, 100);
					})
					.end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.include(v, " -d-swap-view-drag", "Swipe not started after drag threshold");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 100, 100);
					}).end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.include(v, " -d-swap-view-drag", "Missing -d-swap-view-drag class during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[0]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "-100", "Wrong transform during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[1]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "300", "Wrong transform during swipe");
					})
					.releaseMouseButton()
					.sleep(500)
					.execute("return document.getElementById('sv').className;").then(function (v) {
						assert.notInclude(v, " -d-swap-view-drag", "-d-swap-view-drag class left after swipe");
					})
					.execute("return document.getElementById('sv').children[0].style.visibility;")
					.then(function (v) {
						assert.equal(v, "visible", "child 0 should still be visible after swipe 100");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 310, 100);
					})
					.end()
					.pressMouseButton()
					.sleep(50)
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 100, 100);
					})
					.end()
					.sleep(50)
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[0]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "-208", "Wrong transform during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[1]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "188", "Wrong transform during swipe");
					})
					.releaseMouseButton()
					.sleep(500)
					.execute("return document.getElementById('sv').children[0].style.visibility;")
					.then(function (v) {
						assert.equal(v, "hidden", "child 0 should be hidden after swipe > 200");
					})
					.execute("return document.getElementById('sv').children[1].style.visibility;")
					.then(function (v) {
						assert.equal(v, "visible", "child 1 should be visible after swipe > 200");
					})
					;
			}
		},

		"ViewIndicator update": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
				return this.skip("SafariDriver doesn't support moveTo.");
			} else {
				return remote
					.execute("return document.getElementById('vi').children[0].className;")
					.then(function (v) {
						assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 0");
						assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 0");
					})
					.execute("return document.getElementById('vi').children[1].className;")
					.then(function (v) {
						assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 1");
						assert.include(v, "-d-view-indicator-dot-selected", "Extra class for vi child 1");
					})
					.execute("return document.getElementById('vi').children[2].className;")
					.then(function (v) {
						assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 2");
						assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 2");
					})
					;
			}
		},

		"SwapView swipe gesture (left->right)": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
				return this.skip("SafariDriver doesn't support moveTo.");
			} else {
				return remote
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 200, 100);
					})
					.end()
					.pressMouseButton()
					.sleep(50)
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 205, 100);
					})
					.end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.notInclude(v, " -d-swap-view-drag", "Swipe started before drag threshold");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 211, 100);
					})
					.end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.include(v, " -d-swap-view-drag", "Swipe not started after drag threshold");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 300, 100);
					}).end()
					.sleep(50)
					.execute("return document.getElementById('sv').className;")
					.then(function (v) {
						assert.include(v, " -d-swap-view-drag", "Missing -d-swap-view-drag class during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[0]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "-300", "Wrong transform during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[1]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "100", "Wrong transform during swipe");
					})
					.releaseMouseButton()
					.sleep(500)
					.execute("return document.getElementById('sv').className;").then(function (v) {
						assert.notInclude(v, " -d-swap-view-drag", "-d-swap-view-drag class left after swipe");
					})
					.execute("return document.getElementById('sv').children[1].style.visibility;")
					.then(function (v) {
						assert.equal(v, "visible", "child 1 should still be visible after swipe 100");
					})
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 90, 100);
					})
					.end()
					.pressMouseButton()
					.sleep(50)
					.findById("sv")
					.then(function (element) {
						return remote.moveMouseTo(element, 300, 100);
					})
					.end()
					.sleep(50)
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[0]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "-188", "Wrong transform during swipe");
					})
					.execute("var s = window.getComputedStyle(document.getElementById('sv').children[1]);" +
						"return s.webkitTransform || s.transform;")
					.then(function (v) {
						assert.equal(getTransformX(v), "208", "Wrong transform during swipe");
					})
					.releaseMouseButton()
					.sleep(500)
					.execute("return document.getElementById('sv').children[0].style.visibility;")
					.then(function (v) {
						assert.equal(v, "visible", "child 0 should be visible after swipe > 200");
					})
					.execute("return document.getElementById('sv').children[1].style.visibility;")
					.then(function (v) {
						assert.equal(v, "hidden", "child 1 should be hidden after swipe > 200");
					})
					;
			}
		},

		"ViewIndicator click dot": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
				return this.skip("SafariDriver doesn't support moveTo.");
			}
			return remote
				.findById("vi")
				.then(function (element) {
					return remote.moveMouseTo(element, 220, 5);
				})
				.end()
				.pressMouseButton()
				.sleep(50)
				.releaseMouseButton()
				.sleep(500)
				.execute("return document.getElementById('vi').children[0].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 0");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 0");
				})
				.execute("return document.getElementById('vi').children[1].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 1");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 1");
				})
				.execute("return document.getElementById('vi').children[2].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 2");
					assert.include(v, "-d-view-indicator-dot-selected", "Missing class for vi child 2");
				})
				;
		},

		"ViewIndicator click left": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
				return this.skip("SafariDriver doesn't support moveTo.");
			}
			return remote
				.findById("vi")
				.then(function (element) {
					return remote.moveMouseTo(element, 100, 5);
				})
				.end()
				.pressMouseButton()
				.sleep(50)
				.releaseMouseButton()
				.sleep(500)
				.execute("return document.getElementById('vi').children[0].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 0");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 0");
				})
				.execute("return document.getElementById('vi').children[1].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 1");
					assert.include(v, "-d-view-indicator-dot-selected", "Missing class for vi child 1");
				})
				.execute("return document.getElementById('vi').children[2].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 2");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 2");
				})
				;
		},

		"ViewIndicator click right": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName)) {
				// SafariDriver doesn't support moveTo, see https://code.google.com/p/selenium/issues/detail?id=4136
				return this.skip("SafariDriver doesn't support moveTo.");
			}
			return remote
				.findById("vi")
				.then(function (element) {
					return remote.moveMouseTo(element, 300, 5);
				})
				.end()
				.pressMouseButton()
				.sleep(50)
				.releaseMouseButton()
				.sleep(500)
				.execute("return document.getElementById('vi').children[0].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 0");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 0");
				})
				.execute("return document.getElementById('vi').children[1].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 1");
					assert.notInclude(v, "-d-view-indicator-dot-selected", "Extra class for vi child 1");
				})
				.execute("return document.getElementById('vi').children[2].className;")
				.then(function (v) {
					assert.include(v, "-d-view-indicator-dot", "Missing class for vi child 2");
					assert.include(v, "-d-view-indicator-dot-selected", "Missing class for vi child 2");
				})
				;
		},

		"SwapView keyboard accessibility": function () {
			this.timeout = intern.config.TEST_TIMEOUT;
			var remote = this.remote;
			if (/safari|iOS|selendroid/.test(remote.environmentType.browserName) || remote.environmentType.safari) {
				// SafariDriver doesn't support tabbing, see https://code.google.com/p/selenium/issues/detail?id=5403
				// Same problem with selendroid and iOS, apparently
				return this.skip("SafariDriver doesn't support tabbing.");
			}
			return remote
				// keyb nav
				// give the focus to the button to have a ref starting point in the chain
				.execute("return document.getElementById('b1').focus();")
				.getActiveElement()
				.end()
				.sleep(400)
				.pressKeys(keys.TAB) // Press TAB -> cb1
				.sleep(400)
				.getActiveElement()
				.getAttribute("class")
				.then(function (v) {
					assert.include(v, "d-swap-view", "SwapView should be focused after 1st TAB.");
				})
				.end()
				.pressKeys(keys.PAGE_DOWN) // previous -> child 1
				.sleep(500)
				.execute("return document.getElementById('sv').children[1].style.visibility;")
				.then(function (v) {
					assert.equal(v, "visible", "child 1 should be visible after Page Down key");
				})
				.pressKeys(keys.PAGE_UP) // next -> child 2
				.sleep(500)
				.execute("return document.getElementById('sv').children[2].style.visibility;")
				.then(function (v) {
					assert.equal(v, "visible", "child 2 should be visible after Page Up key");
				})
				.pressKeys(keys.TAB) // Press TAB -> End button
				.sleep(400)
				.getActiveElement()
				.getVisibleText()
				.then(function (v) {
					assert.equal(v, "End", "Unexpected focused element after 2nd TAB.");
				})
				.end()
				;
		}
	});
});