<commit_before>/*$Id$
* $Revision$
* $Date$
* $Author$
*
* The Netarchive Suite - Software to harvest and preserve websites
* Copyright 2004-2010 <NAME> and Statsbiblioteket, Denmark
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/
package dk.netarkivet.archive.arcrepository.distribute;

import java.io.File;
import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import dk.netarkivet.archive.arcrepository.bitpreservation.AdminDataMessage;
import dk.netarkivet.archive.bitarchive.distribute.BatchMessage;
import dk.netarkivet.archive.bitarchive.distribute.BatchReplyMessage;
import dk.netarkivet.archive.bitarchive.distribute.GetFileMessage;
import dk.netarkivet.archive.bitarchive.distribute.GetMessage;
import dk.netarkivet.archive.bitarchive.distribute.RemoveAndGetFileMessage;
import dk.netarkivet.archive.checksum.distribute.CorrectMessage;
import dk.netarkivet.archive.checksum.distribute.GetAllChecksumsMessage;
import dk.netarkivet.archive.checksum.distribute.GetAllFilenamesMessage;
import dk.netarkivet.archive.checksum.distribute.GetChecksumMessage;
import dk.netarkivet.common.distribute.ChannelID;
import dk.netarkivet.common.distribute.Channels;
import dk.netarkivet.common.distribute.JMSConnectionFactory;
import dk.netarkivet.common.distribute.NetarkivetMessage;
import dk.netarkivet.common.distribute.RemoteFile;
import dk.netarkivet.common.distribute.RemoteFileFactory;
import dk.netarkivet.common.distribute.Synchronizer;
import dk.netarkivet.common.distribute.arcrepository.ArcRepositoryClient;
import dk.netarkivet.common.distribute.arcrepository.BatchStatus;
import dk.netarkivet.common.distribute.arcrepository.ReplicaStoreState;
import dk.netarkivet.common.distribute.arcrepository.BitarchiveRecord;
import dk.netarkivet.common.distribute.arcrepository.Replica;
import dk.netarkivet.common.exceptions.ArgumentNotValid;
import dk.netarkivet.common.exceptions.IOFailure;
import dk.netarkivet.common.utils.ExceptionUtils;
import dk.netarkivet.common.utils.FileUtils;
import dk.netarkivet.common.utils.NotificationsFactory;
import dk.netarkivet.common.utils.Settings;
import dk.netarkivet.common.utils.batch.FileBatchJob;

/**
 * Client side usage of an arc repository. All requests are forwarded to the
 * ArcRepositoryServer over the network. get and store messages are retried a
 * number of time before giving up, and will timeout after a specified time.
 */
public class JMSArcRepositoryClient extends Synchronizer implements
                                                         ArcRepositoryClient {
    /** The default place in classpath where the settings file can be found. */
    private static String defaultSettingsClasspath
            = "dk/netarkivet/archive/arcrepository/distribute/"
                +"JMSArcRepositoryClientSettings.xml";

    /*
     * The static initialiser is called when the class is loaded.
     * It will add default values for all settings defined in this class, by
     * loading them from a settings.xml file in classpath.
     */
    static {
        Settings.addDefaultClasspathSettings(defaultSettingsClasspath);
    }
    
    /** The amount of milliseconds in a second. 1000. */
    private static final int MILLISECONDS_PER_SECOND = 1000;

    /** the one and only JMSArcRepositoryClient instance. */
    private static JMSArcRepositoryClient instance;

    /** Logging output place. */
    protected final Log log = LogFactory.getLog(getClass());

    /** Listens on this queue for replies. */
    private final ChannelID replyQ;

    /** The number of times to try sending a store message before giving up. */
    private long storeRetries;

    /** The length of time to wait for a store reply before giving up. */
    private long storeTimeout;

    /** The length of time to wait for a get reply before giving up. */
    private long getTimeout;

    // NOTE: The constants defining setting names below are left non-final on
    // purpose! Otherwise, the static initialiser that loads default values
    // will not run.

    /**
     * <b>settings.common.arcrepositoryClient.getTimeout</b>: <br> The setting
     * for how many milliseconds we will wait before giving up on a lookup
     * request to the Arcrepository.
     */
    public static final String ARCREPOSITORY_GET_TIMEOUT
            = "settings.common.arcrepositoryClient.getTimeout";

    /**
     * <b>settings.common.arcrepositoryClient.storeRetries</b>: <br> The setting
     * for the number of times to try sending a store message before failing,
     * including the first attempt.
     */
    public static final String ARCREPOSITORY_STORE_RETRIES
            = "settings.common.arcrepositoryClient.storeRetries";

    /**
     * <b>settings.common.arcrepositoryClient.storeTimeout</b>: <br> the setting
     * for the timeout in milliseconds before retrying when calling {@link
     * ArcRepositoryClient#store(File)}.
     */
    public static final String ARCREPOSITORY_STORE_TIMEOUT
            = "settings.common.arcrepositoryClient.storeTimeout";

    /** Adds this Synchronizer as listener on a jms connection. */
    protected JMSArcRepositoryClient() {
        storeRetries = Settings.getLong(
                ARCREPOSITORY_STORE_RETRIES);
        storeTimeout = Settings.getLong(
                ARCREPOSITORY_STORE_TIMEOUT);
        getTimeout = Settings.getLong(ARCREPOSITORY_GET_TIMEOUT);

        log.info("JMSArcRepositoryClient will retry a store " + storeRetries
                 + " times and timeout on each try after " + storeTimeout
                 + " milliseconds, and timeout on each getrequest after "
                 + getTimeout + " milliseconds.");
        replyQ = Channels.getThisReposClient();
        JMSConnectionFactory.getInstance().setListener(replyQ, this);
        log.info("JMSArcRepository listens for replies on channel '"
                 + replyQ + "'");
    }

    /**
     * Get an JMSArcRepositoryClient instance. This is guaranteed to be a
     * singleton.
     *
     * @return an JMSArcRepositoryClient instance.
     */
    public static synchronized JMSArcRepositoryClient getInstance() {
        if (instance == null) {
            instance = new JMSArcRepositoryClient();
        }
        return instance;
    }

    /** Removes this object as a JMS listener. */
    public void close() {
        synchronized (JMSArcRepositoryClient.class) {
            JMSConnectionFactory.getInstance().removeListener(replyQ, this);
            instance = null;
        }
    }

    /**
     * Sends a GetMessage on the "TheArcrepos" queue and waits for a reply. This
     * is a blocking call. Returns null if no message is returned within
     * Settings.ARCREPOSITORY_GET_TIMEOUT
     *
     * @param arcfile The name of a file.
     * @param index   The offset of the wanted record in the file
     *
     * @return a BitarchiveRecord-object or null if request times out or object
     *         is not found.
     *
     * @throws ArgumentNotValid If the given arcfile is null or empty, or the
     *                          given index is negative.
     * @throws IOFailure        If a wrong message is returned or the get
     *                          operation failed.
     */
    public BitarchiveRecord get(String arcfile, long index)
            throws ArgumentNotValid, IOFailure {
        ArgumentNotValid.checkNotNullOrEmpty(arcfile, "arcfile");
        ArgumentNotValid.checkNotNegative(index, "index");
        log.debug("Requesting get of record '" + arcfile + ":" + index + "'");
        long start = System.currentTimeMillis();
        GetMessage requestGetMsg = new GetMessage(Channels.getTheRepos(),
                                                  replyQ, arcfile, index);
        NetarkivetMessage replyNetMsg = sendAndWaitForOneReply(requestGetMsg,
                                                               getTimeout);
        long timePassed = System.currentTimeMillis() - start;
        log.debug("Reply received after " + (timePassed 
                / MILLISECONDS_PER_SECOND) + " seconds");
        if (replyNetMsg == null) {
            log.info("Request for record(" + arcfile + ":" + index
                     + ") timed out after "
                     + (getTimeout / MILLISECONDS_PER_SECOND)
                     + " seconds. Returning null BitarchiveRecord");
            return null;
        }
        GetMessage replyGetMsg;
        try {
            replyGetMsg = (GetMessage) replyNetMsg;
        } catch (ClassCastException e) {
            throw new IOFailure("Received invalid argument reply: '"
                    + replyNetMsg + "'", e);
        }
        if (!replyGetMsg.isOk()) {
            throw new IOFailure("GetMessage failed: '"
                    + replyGetMsg.getErrMsg() + "'");
        }
        return replyGetMsg.getRecord();
    }

    /**
     * Synchronously retrieves a file from a bitarchive and places it in a local
     * file. This is the interface for sending GetFileMessage on the
     * "TheArcrepos" queue. This is a blocking call.
     *
     * @param arcfilename Name of the arcfile to retrieve.
     * @param replica     The bitarchive to retrieve the data from.
     * @param toFile      Filename of a place where the file fetched can be
     *                    put.
     * @throws ArgumentNotValid If the arcfilename are null or empty, or if 
     * either replica or toFile is null.
     * @throws IOFailure if there are problems getting a reply or the file could
     *                   not be found.
     */
    public void getFile(String arcfilename, Replica replica, File toFile) 
        throws ArgumentNotValid, IOFailure {
        ArgumentNotValid.checkNotNullOrEmpty(arcfilename, "arcfilename");
        ArgumentNotValid.checkNotNull(replica, "replica");
        ArgumentNotValid.checkNotNull(toFile, "toFile");

        log.debug("Requesting get of file '" + arcfilename + "' from '"
                  + replica + "'");
        //ArgumentNotValid.checkNotNull(replyQ, "replyQ must not be null");
        GetFileMessage gfMsg = new GetFileMessage(Channels.getTheRepos(),
                                                  replyQ, arcfilename,
                                                  replica.getId());
        GetFileMessage getFileMessage
                = (GetFileMessage) sendAndWaitForOneReply(gfMsg, 0);
        if (getFileMessage == null) {
            throw new IOFailure("GetFileMessage timed out before returning."
                    + "File not found?");
        } else if (!getFileMessage.isOk()) {
            throw new IOFailure("GetFileMessage failed: " 
                    + getFileMessage.getErrMsg());
        } else {
            getFileMessage.getData(toFile);
        }
    }

    /**
     * Sends a StoreMessage via the synchronized JMS connection method
     * sendAndWaitForOneReply(). After a successful storage operation, both the
     * local copy of the file and the copy on the ftp server are deleted.
     *
     * @param file A file to be stored. Must exist.
     *
     * @throws IOFailure        thrown if store is unsuccessful, or failed to
     *                          clean up files locally or on the ftp server
     *                          after the store operation.
     * @throws ArgumentNotValid if file parameter is null or file is not an
     *                          existing file.
     */
    public void store(File file) throws IOFailure, ArgumentNotValid {
        ArgumentNotValid.checkNotNull(file, "file");
        ArgumentNotValid.checkTrue(file.isFile(), "The file '" + file.getPath()
                + "' is not an existing file.");

        StringBuilder messages = new StringBuilder();
        for (long i = 0; i < storeRetries; i++) {
            StoreMessage outMsg = null;
            try {
                log.debug("Sending a StoreMessage with file '" + file.getPath()
                          + "'");
                outMsg = new StoreMessage(replyQ, file);
                NetarkivetMessage replyMsg
                        = sendAndWaitForOneReply(outMsg, storeTimeout);
                if (replyMsg != null && replyMsg.isOk()) {
                    try {
                        FileUtils.removeRecursively(file);
                    } catch (IOFailure e) {
                        log.warn("Failed to clean up '"
                                 + file.getAbsolutePath() + "'", e);
                        // Not fatal
                    }
                    return;
                } else if (replyMsg == null) {
                    String msg = "Timed out"
                                 + " while waiting for reply on store of file '"
                                 + file.getPath() + "' on attempt number "
                                 + (i + 1)
                                 + " of " + storeRetries;
                    log.warn(msg);
                    messages.append(msg).append("\n");
                } else {
                    String msg = "The returned message '" + replyMsg
                                 + "' was not ok"
                                 + " while waiting for reply on store of file '"
                                 + file.getPath() + "' on attempt number "
                                 + (i + 1)
                                 + " of " + storeRetries
                                 + ". Error message was '"
                                 + replyMsg.getErrMsg() + "'";
                    log.warn(msg);
                    messages.append(msg).append("\n");
                }
            } catch (Exception e) {
                String msg = "Client-side exception occurred while storing '"
                             + file.getPath() + "' on attempt number " + (i + 1)
                             + " of " + storeRetries + ".";
                log.warn(msg, e);
                messages.append(msg).append("\n");
                messages.append(ExceptionUtils.getStackTrace(e));
            } finally {
                if (outMsg != null) {
                    cleanUpAfterStore(outMsg);
                }
            }
        }
        String errMsg = "Could not store '" + file.getPath() + "' after "
                        + storeRetries + " attempts. Giving up.\n" + messages;
        log.warn(errMsg);
        NotificationsFactory.getInstance().errorEvent(errMsg);
        throw new IOFailure(errMsg);
    }

    /**
     * Tries to clean up a file on the FTP server after a store operation. Will
     * not throw exception on error, merely log exception.
     *
     * @param m the StoreMessage sent back as reply
     */
    private void cleanUpAfterStore(StoreMessage m) {
        RemoteFile rf;
        try {
            rf = m.getRemoteFile();
        } catch (Exception e) {
            log.warn("Could not get remote file object from message " + m, e);
            return;
        }
        try {
            rf.cleanup();
        } catch (Exception e) {
            log.warn("Could not delete remote file on ftp server: " + rf, e);
        }
    }
    
    /**
     * Runs a batch batch job on each file in the ArcRepository.
     * 
     * Note: The id for the batchjob is the empty string, which removes the 
     * possibility of terminating the batchjob remotely while it is running.
     *
     * @param job An object that implements the FileBatchJob interface. The
     *  initialize() method will be called before processing and the finish()
     *  method will be called afterwards. The process() method will be called
     *  with each File entry. An optional function postProcess() allows handling
     *  the combined results of the batchjob, e.g. summing the results, sorting,
     *  etc.
     *
     * @param replicaId The archive to execute the job on.
     * @param args The arguments for the batchjob.
     * @return The status of the batch job after it ended.
     */
    public BatchStatus batch(FileBatchJob job, String replicaId, 
            String... args) {
        return batch(job, replicaId, "", args);
    }

    /**
     * Runs a batch job on each file in the ArcRepository.
     *
     * @param job An object that implements the FileBatchJob interface. The
     *  initialize() method will be called before processing and the finish()
     *  method will be called afterwards. The process() method will be called
     *  with each File entry. An optional function postProcess() allows handling
     *  the combined results of the batchjob, e.g. summing the results, sorting,
     *  etc.
     *
     * @param replicaId The archive to execute the job on.
<<<<<<< .mine
     * @param args The arguments for the batchjob.
     * @param batchId The id for the batch process.
     * @return The status of the batch job after it ended.
=======
     * @param args The arguments for the batchjob. This is allowed to be null.
     * @return The status of the batch job after it has ended.
>>>>>>> .r1542
     * @throws ArgumentNotValid If the job is null or the replicaId is either 
     * null or the empty string.
     * @throws IOFailure If no result file is returned.
     */
    public BatchStatus batch(FileBatchJob job, String replicaId, String batchId,
            String... args) throws IOFailure, ArgumentNotValid {
        ArgumentNotValid.checkNotNull(job, "FileBatchJob job");
        ArgumentNotValid.checkNotNullOrEmpty(replicaId, "String replicaId");

        log.debug("Starting batchjob '" + job + "' running on replica '"
                  + replicaId + "'");
        BatchMessage bMsg = new BatchMessage(Channels.getTheRepos(), replyQ,
                job, replicaId, batchId, args);
        log.debug("Sending batchmessage to queue '" + Channels.getTheRepos()
                  + "' with replyqueue set to '" + replyQ + "'");
        BatchReplyMessage brMsg =
                (BatchReplyMessage) sendAndWaitForOneReply(bMsg, 0);
        if (!brMsg.isOk()) {
            String msg = "The batch job '" + bMsg
                         + "' resulted in the following "
                         + "error: " + brMsg.getErrMsg();
            log.warn(msg);
            if (brMsg.getResultFile() == null) {
                // If no result is available at all, this is non-recoverable
                throw new IOFailure(msg);<commit_msg>Fixed duplicate javadoc inserted after merging code.<commit_after>/*$Id$
* $Revision$
* $Date$
* $Author$
*
* The Netarchive Suite - Software to harvest and preserve websites
* Copyright 2004-2010 <NAME> and Statsbiblioteket, Denmark
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/
package dk.netarkivet.archive.arcrepository.distribute;

import java.io.File;
import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import dk.netarkivet.archive.arcrepository.bitpreservation.AdminDataMessage;
import dk.netarkivet.archive.bitarchive.distribute.BatchMessage;
import dk.netarkivet.archive.bitarchive.distribute.BatchReplyMessage;
import dk.netarkivet.archive.bitarchive.distribute.GetFileMessage;
import dk.netarkivet.archive.bitarchive.distribute.GetMessage;
import dk.netarkivet.archive.bitarchive.distribute.RemoveAndGetFileMessage;
import dk.netarkivet.archive.checksum.distribute.CorrectMessage;
import dk.netarkivet.archive.checksum.distribute.GetAllChecksumsMessage;
import dk.netarkivet.archive.checksum.distribute.GetAllFilenamesMessage;
import dk.netarkivet.archive.checksum.distribute.GetChecksumMessage;
import dk.netarkivet.common.distribute.ChannelID;
import dk.netarkivet.common.distribute.Channels;
import dk.netarkivet.common.distribute.JMSConnectionFactory;
import dk.netarkivet.common.distribute.NetarkivetMessage;
import dk.netarkivet.common.distribute.RemoteFile;
import dk.netarkivet.common.distribute.RemoteFileFactory;
import dk.netarkivet.common.distribute.Synchronizer;
import dk.netarkivet.common.distribute.arcrepository.ArcRepositoryClient;
import dk.netarkivet.common.distribute.arcrepository.BatchStatus;
import dk.netarkivet.common.distribute.arcrepository.ReplicaStoreState;
import dk.netarkivet.common.distribute.arcrepository.BitarchiveRecord;
import dk.netarkivet.common.distribute.arcrepository.Replica;
import dk.netarkivet.common.exceptions.ArgumentNotValid;
import dk.netarkivet.common.exceptions.IOFailure;
import dk.netarkivet.common.utils.ExceptionUtils;
import dk.netarkivet.common.utils.FileUtils;
import dk.netarkivet.common.utils.NotificationsFactory;
import dk.netarkivet.common.utils.Settings;
import dk.netarkivet.common.utils.batch.FileBatchJob;

/**
 * Client side usage of an arc repository. All requests are forwarded to the
 * ArcRepositoryServer over the network. get and store messages are retried a
 * number of time before giving up, and will timeout after a specified time.
 */
public class JMSArcRepositoryClient extends Synchronizer implements
                                                         ArcRepositoryClient {
    /** The default place in classpath where the settings file can be found. */
    private static String defaultSettingsClasspath
            = "dk/netarkivet/archive/arcrepository/distribute/"
                +"JMSArcRepositoryClientSettings.xml";

    /*
     * The static initialiser is called when the class is loaded.
     * It will add default values for all settings defined in this class, by
     * loading them from a settings.xml file in classpath.
     */
    static {
        Settings.addDefaultClasspathSettings(defaultSettingsClasspath);
    }
    
    /** The amount of milliseconds in a second. 1000. */
    private static final int MILLISECONDS_PER_SECOND = 1000;

    /** the one and only JMSArcRepositoryClient instance. */
    private static JMSArcRepositoryClient instance;

    /** Logging output place. */
    protected final Log log = LogFactory.getLog(getClass());

    /** Listens on this queue for replies. */
    private final ChannelID replyQ;

    /** The number of times to try sending a store message before giving up. */
    private long storeRetries;

    /** The length of time to wait for a store reply before giving up. */
    private long storeTimeout;

    /** The length of time to wait for a get reply before giving up. */
    private long getTimeout;

    // NOTE: The constants defining setting names below are left non-final on
    // purpose! Otherwise, the static initialiser that loads default values
    // will not run.

    /**
     * <b>settings.common.arcrepositoryClient.getTimeout</b>: <br> The setting
     * for how many milliseconds we will wait before giving up on a lookup
     * request to the Arcrepository.
     */
    public static final String ARCREPOSITORY_GET_TIMEOUT
            = "settings.common.arcrepositoryClient.getTimeout";

    /**
     * <b>settings.common.arcrepositoryClient.storeRetries</b>: <br> The setting
     * for the number of times to try sending a store message before failing,
     * including the first attempt.
     */
    public static final String ARCREPOSITORY_STORE_RETRIES
            = "settings.common.arcrepositoryClient.storeRetries";

    /**
     * <b>settings.common.arcrepositoryClient.storeTimeout</b>: <br> the setting
     * for the timeout in milliseconds before retrying when calling {@link
     * ArcRepositoryClient#store(File)}.
     */
    public static final String ARCREPOSITORY_STORE_TIMEOUT
            = "settings.common.arcrepositoryClient.storeTimeout";

    /** Adds this Synchronizer as listener on a jms connection. */
    protected JMSArcRepositoryClient() {
        storeRetries = Settings.getLong(
                ARCREPOSITORY_STORE_RETRIES);
        storeTimeout = Settings.getLong(
                ARCREPOSITORY_STORE_TIMEOUT);
        getTimeout = Settings.getLong(ARCREPOSITORY_GET_TIMEOUT);

        log.info("JMSArcRepositoryClient will retry a store " + storeRetries
                 + " times and timeout on each try after " + storeTimeout
                 + " milliseconds, and timeout on each getrequest after "
                 + getTimeout + " milliseconds.");
        replyQ = Channels.getThisReposClient();
        JMSConnectionFactory.getInstance().setListener(replyQ, this);
        log.info("JMSArcRepository listens for replies on channel '"
                 + replyQ + "'");
    }

    /**
     * Get an JMSArcRepositoryClient instance. This is guaranteed to be a
     * singleton.
     *
     * @return an JMSArcRepositoryClient instance.
     */
    public static synchronized JMSArcRepositoryClient getInstance() {
        if (instance == null) {
            instance = new JMSArcRepositoryClient();
        }
        return instance;
    }

    /** Removes this object as a JMS listener. */
    public void close() {
        synchronized (JMSArcRepositoryClient.class) {
            JMSConnectionFactory.getInstance().removeListener(replyQ, this);
            instance = null;
        }
    }

    /**
     * Sends a GetMessage on the "TheArcrepos" queue and waits for a reply. This
     * is a blocking call. Returns null if no message is returned within
     * Settings.ARCREPOSITORY_GET_TIMEOUT
     *
     * @param arcfile The name of a file.
     * @param index   The offset of the wanted record in the file
     *
     * @return a BitarchiveRecord-object or null if request times out or object
     *         is not found.
     *
     * @throws ArgumentNotValid If the given arcfile is null or empty, or the
     *                          given index is negative.
     * @throws IOFailure        If a wrong message is returned or the get
     *                          operation failed.
     */
    public BitarchiveRecord get(String arcfile, long index)
            throws ArgumentNotValid, IOFailure {
        ArgumentNotValid.checkNotNullOrEmpty(arcfile, "arcfile");
        ArgumentNotValid.checkNotNegative(index, "index");
        log.debug("Requesting get of record '" + arcfile + ":" + index + "'");
        long start = System.currentTimeMillis();
        GetMessage requestGetMsg = new GetMessage(Channels.getTheRepos(),
                                                  replyQ, arcfile, index);
        NetarkivetMessage replyNetMsg = sendAndWaitForOneReply(requestGetMsg,
                                                               getTimeout);
        long timePassed = System.currentTimeMillis() - start;
        log.debug("Reply received after " + (timePassed 
                / MILLISECONDS_PER_SECOND) + " seconds");
        if (replyNetMsg == null) {
            log.info("Request for record(" + arcfile + ":" + index
                     + ") timed out after "
                     + (getTimeout / MILLISECONDS_PER_SECOND)
                     + " seconds. Returning null BitarchiveRecord");
            return null;
        }
        GetMessage replyGetMsg;
        try {
            replyGetMsg = (GetMessage) replyNetMsg;
        } catch (ClassCastException e) {
            throw new IOFailure("Received invalid argument reply: '"
                    + replyNetMsg + "'", e);
        }
        if (!replyGetMsg.isOk()) {
            throw new IOFailure("GetMessage failed: '"
                    + replyGetMsg.getErrMsg() + "'");
        }
        return replyGetMsg.getRecord();
    }

    /**
     * Synchronously retrieves a file from a bitarchive and places it in a local
     * file. This is the interface for sending GetFileMessage on the
     * "TheArcrepos" queue. This is a blocking call.
     *
     * @param arcfilename Name of the arcfile to retrieve.
     * @param replica     The bitarchive to retrieve the data from.
     * @param toFile      Filename of a place where the file fetched can be
     *                    put.
     * @throws ArgumentNotValid If the arcfilename are null or empty, or if 
     * either replica or toFile is null.
     * @throws IOFailure if there are problems getting a reply or the file could
     *                   not be found.
     */
    public void getFile(String arcfilename, Replica replica, File toFile) 
        throws ArgumentNotValid, IOFailure {
        ArgumentNotValid.checkNotNullOrEmpty(arcfilename, "arcfilename");
        ArgumentNotValid.checkNotNull(replica, "replica");
        ArgumentNotValid.checkNotNull(toFile, "toFile");

        log.debug("Requesting get of file '" + arcfilename + "' from '"
                  + replica + "'");
        //ArgumentNotValid.checkNotNull(replyQ, "replyQ must not be null");
        GetFileMessage gfMsg = new GetFileMessage(Channels.getTheRepos(),
                                                  replyQ, arcfilename,
                                                  replica.getId());
        GetFileMessage getFileMessage
                = (GetFileMessage) sendAndWaitForOneReply(gfMsg, 0);
        if (getFileMessage == null) {
            throw new IOFailure("GetFileMessage timed out before returning."
                    + "File not found?");
        } else if (!getFileMessage.isOk()) {
            throw new IOFailure("GetFileMessage failed: " 
                    + getFileMessage.getErrMsg());
        } else {
            getFileMessage.getData(toFile);
        }
    }

    /**
     * Sends a StoreMessage via the synchronized JMS connection method
     * sendAndWaitForOneReply(). After a successful storage operation, both the
     * local copy of the file and the copy on the ftp server are deleted.
     *
     * @param file A file to be stored. Must exist.
     *
     * @throws IOFailure        thrown if store is unsuccessful, or failed to
     *                          clean up files locally or on the ftp server
     *                          after the store operation.
     * @throws ArgumentNotValid if file parameter is null or file is not an
     *                          existing file.
     */
    public void store(File file) throws IOFailure, ArgumentNotValid {
        ArgumentNotValid.checkNotNull(file, "file");
        ArgumentNotValid.checkTrue(file.isFile(), "The file '" + file.getPath()
                + "' is not an existing file.");

        StringBuilder messages = new StringBuilder();
        for (long i = 0; i < storeRetries; i++) {
            StoreMessage outMsg = null;
            try {
                log.debug("Sending a StoreMessage with file '" + file.getPath()
                          + "'");
                outMsg = new StoreMessage(replyQ, file);
                NetarkivetMessage replyMsg
                        = sendAndWaitForOneReply(outMsg, storeTimeout);
                if (replyMsg != null && replyMsg.isOk()) {
                    try {
                        FileUtils.removeRecursively(file);
                    } catch (IOFailure e) {
                        log.warn("Failed to clean up '"
                                 + file.getAbsolutePath() + "'", e);
                        // Not fatal
                    }
                    return;
                } else if (replyMsg == null) {
                    String msg = "Timed out"
                                 + " while waiting for reply on store of file '"
                                 + file.getPath() + "' on attempt number "
                                 + (i + 1)
                                 + " of " + storeRetries;
                    log.warn(msg);
                    messages.append(msg).append("\n");
                } else {
                    String msg = "The returned message '" + replyMsg
                                 + "' was not ok"
                                 + " while waiting for reply on store of file '"
                                 + file.getPath() + "' on attempt number "
                                 + (i + 1)
                                 + " of " + storeRetries
                                 + ". Error message was '"
                                 + replyMsg.getErrMsg() + "'";
                    log.warn(msg);
                    messages.append(msg).append("\n");
                }
            } catch (Exception e) {
                String msg = "Client-side exception occurred while storing '"
                             + file.getPath() + "' on attempt number " + (i + 1)
                             + " of " + storeRetries + ".";
                log.warn(msg, e);
                messages.append(msg).append("\n");
                messages.append(ExceptionUtils.getStackTrace(e));
            } finally {
                if (outMsg != null) {
                    cleanUpAfterStore(outMsg);
                }
            }
        }
        String errMsg = "Could not store '" + file.getPath() + "' after "
                        + storeRetries + " attempts. Giving up.\n" + messages;
        log.warn(errMsg);
        NotificationsFactory.getInstance().errorEvent(errMsg);
        throw new IOFailure(errMsg);
    }

    /**
     * Tries to clean up a file on the FTP server after a store operation. Will
     * not throw exception on error, merely log exception.
     *
     * @param m the StoreMessage sent back as reply
     */
    private void cleanUpAfterStore(StoreMessage m) {
        RemoteFile rf;
        try {
            rf = m.getRemoteFile();
        } catch (Exception e) {
            log.warn("Could not get remote file object from message " + m, e);
            return;
        }
        try {
            rf.cleanup();
        } catch (Exception e) {
            log.warn("Could not delete remote file on ftp server: " + rf, e);
        }
    }
    
    /**
     * Runs a batch batch job on each file in the ArcRepository.
     * 
     * Note: The id for the batchjob is the empty string, which removes the 
     * possibility of terminating the batchjob remotely while it is running.
     *
     * @param job An object that implements the FileBatchJob interface. The
     *  initialize() method will be called before processing and the finish()
     *  method will be called afterwards. The process() method will be called
     *  with each File entry. An optional function postProcess() allows handling
     *  the combined results of the batchjob, e.g. summing the results, sorting,
     *  etc.
     *
     * @param replicaId The archive to execute the job on.
     * @param args The arguments for the batchjob.
     * @return The status of the batch job after it ended.
     */
    public BatchStatus batch(FileBatchJob job, String replicaId, 
            String... args) {
        return batch(job, replicaId, "", args);
    }

    /**
     * Runs a batch job on each file in the ArcRepository.
     *
     * @param job An object that implements the FileBatchJob interface. The
     *  initialize() method will be called before processing and the finish()
     *  method will be called afterwards. The process() method will be called
     *  with each File entry. An optional function postProcess() allows handling
     *  the combined results of the batchjob, e.g. summing the results, sorting,
     *  etc.
     *
     * @param replicaId The archive to execute the job on.
     * @param args The arguments for the batchjob. This is allowed to be null.
     * @param batchId The id for the batch process.
     * @return The status of the batch job after it ended.
     * @throws ArgumentNotValid If the job is null or the replicaId is either 
     * null or the empty string.
     * @throws IOFailure If no result file is returned.
     */
    public BatchStatus batch(FileBatchJob job, String replicaId, String batchId,
            String... args) throws IOFailure, ArgumentNotValid {
        ArgumentNotValid.checkNotNull(job, "FileBatchJob job");
        ArgumentNotValid.checkNotNullOrEmpty(replicaId, "String replicaId");

        log.debug("Starting batchjob '" + job + "' running on replica '"
                  + replicaId + "'");
        BatchMessage bMsg = new BatchMessage(Channels.getTheRepos(), replyQ,
                job, replicaId, batchId, args);
        log.debug("Sending batchmessage to queue '" + Channels.getTheRepos()
                  + "' with replyqueue set to '" + replyQ + "'");
        BatchReplyMessage brMsg =
                (BatchReplyMessage) sendAndWaitForOneReply(bMsg, 0);
        if (!brMsg.isOk()) {
            String msg = "The batch job '" + bMsg
                         + "' resulted in the following "
                         + "error: " + brMsg.getErrMsg();
            log.warn(msg);
            if (brMsg.getResultFile() == null) {
                // If no result is available at all, this is non-recoverable
                throw new IOFailure(msg);
{'fraction_non_alphanumeric': 0.08318359581439594, 'fraction_numerical': 0.0019553958355353557, 'mean_word_length': 2.116610392818908, 'pattern_counts': {'":': 4, '<': 32, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 50, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6892256', 'n_tokens_mistral': 9885, 'n_tokens_neox': 9674, 'n_words': 3151}