class LowDiskPauseProcessor extends Processor {

    @SuppressWarnings("unused")
    private static final long serialVersionUID = 3L;

    /**
     * Logger.
     */
    private static final Logger logger =
        Logger.getLogger(LowDiskPauseProcessor.class.getName());


    protected CrawlController controller;
    public CrawlController getCrawlController() {
        return this.controller;
    }
    @Autowired
    public void setCrawlController(CrawlController controller) {
        this.controller = controller;
    }
    
    /**
     * List of filessystem mounts whose 'available' space should be monitored
     * via 'df' (if available).
     */
    protected List<String> monitorMounts = new ArrayList<String>();
    public List<String> getMonitorMounts() {
        return this.monitorMounts;
    }
    public void setMonitorMounts(List<String> monitorMounts) {
        this.monitorMounts = monitorMounts;
    }

    /**
     * When available space on any monitored mounts falls below this threshold,
     * the crawl will be paused.
     */
    protected int pauseThresholdKb = 500*1024; // 500MB 
    public int getPauseThresholdKb() {
        return this.pauseThresholdKb;
    }
    public void setPauseThresholdKb(int pauseThresholdKb) {
        this.pauseThresholdKb = pauseThresholdKb;
    }
    
    /**
     * Available space via 'df' is rechecked after every increment of this much
     * content (uncompressed) is observed.
     */
    protected int recheckThresholdKb = 200*1024; // 200MB 
    public int getRecheckThresholdKb() {
        return this.recheckThresholdKb;
    }
    public void setRecheckThresholdKb(int recheckThresholdKb) {
        this.recheckThresholdKb = recheckThresholdKb;
    }
    
    protected int contentSinceCheck = 0;
    
    public static final Pattern VALID_DF_OUTPUT = 
        Pattern.compile("(?s)^Filesystem\\s+1K-blocks\\s+Used\\s+Available\\s+Use%\\s+Mounted on\\n.*");
    public static final Pattern AVAILABLE_EXTRACTOR = 
        Pattern.compile("(?m)\\s(\\d+)\\s+\\d+%\\s+(\\S+)$");
    
    public LowDiskPauseProcessor() {
    } 
    
    
    @Override
    protected boolean shouldProcess(CrawlURI curi) {
        return true;
    }

    @Override
    protected void innerProcess(CrawlURI uri) {
        throw new AssertionError();
    }
    
    /**
     * Notes a CrawlURI's content size in its running tally. If the 
     * recheck increment of content has passed through since the last
     * available-space check, checks available space and pauses the 
     * crawl if any monitored mounts are below the configured threshold. 
     * 
     * @param curi CrawlURI to process.
     */
    @Override
    protected ProcessResult innerProcessResult(CrawlURI curi) {
        synchronized (this) {
            contentSinceCheck += curi.getContentSize();
            if (contentSinceCheck/1024 > getRecheckThresholdKb()) {
                ProcessResult r = checkAvailableSpace(curi);
                contentSinceCheck = 0;
                return r;
            } else {
                return ProcessResult.PROCEED;
            }
        }
    }


    /**
     * Probe via 'df' to see if monitored mounts have fallen
     * below the pause available threshold. If so, request a 
     * crawl pause. 
     * @param curi Current context.
     */
    private ProcessResult checkAvailableSpace(CrawlURI curi) {<FILL_FUNCTION_BODY>}
}

        try {
            String df = IOUtils.toString(Runtime.getRuntime().exec(
                    "df -k").getInputStream());
            Matcher matcher = VALID_DF_OUTPUT.matcher(df);
            if(!matcher.matches()) {
                logger.severe("'df -k' output unacceptable for low-disk checking");
                return ProcessResult.PROCEED;
            }
            List<String> monitoredMounts = getMonitorMounts();
            matcher = AVAILABLE_EXTRACTOR.matcher(df);
            while (matcher.find()) {
                String mount = matcher.group(2);
                if (monitoredMounts.contains(mount)) {
                    long availKilobytes = Long.parseLong(matcher.group(1));
                    int thresholdKilobytes = getPauseThresholdKb();
                    if (availKilobytes < thresholdKilobytes ) {
                        logger.log(Level.SEVERE, "Low Disk Pause",
                                availKilobytes + "K available on " + mount
                                        + " (below threshold "
                                        + thresholdKilobytes + "K)");
                        controller.requestCrawlPause();
                        return ProcessResult.PROCEED;
                    }
                }
            }
        } catch (IOException e) {
            curi.getNonFatalFailures().add(e);
        }
        return ProcessResult.PROCEED;
    
<methods>public non-sealed void <init>() ,public void doCheckpoint(org.archive.checkpointing.Checkpoint) throws java.io.IOException,public void finishCheckpoint(org.archive.checkpointing.Checkpoint) ,public static java.lang.String flattenVia(org.archive.modules.CrawlURI) ,public java.lang.String getBeanName() ,public boolean getEnabled() ,public org.archive.spring.KeyedProperties getKeyedProperties() ,public static long getRecordedSize(org.archive.modules.CrawlURI) ,public org.archive.modules.deciderules.DecideRule getShouldProcessRule() ,public long getURICount() ,public static boolean hasHttpAuthenticationCredential(org.archive.modules.CrawlURI) ,public boolean isRunning() ,public static boolean isSuccess(org.archive.modules.CrawlURI) ,public org.archive.modules.ProcessResult process(org.archive.modules.CrawlURI) throws java.lang.InterruptedException,public java.lang.String report() ,public void setBeanName(java.lang.String) ,public void setEnabled(boolean) ,public void setRecoveryCheckpoint(org.archive.checkpointing.Checkpoint) ,public void setShouldProcessRule(org.archive.modules.deciderules.DecideRule) ,public void start() ,public void startCheckpoint(org.archive.checkpointing.Checkpoint) ,public void stop() <variables>protected java.lang.String beanName,protected boolean isRunning,protected org.archive.spring.KeyedProperties kp,protected org.archive.checkpointing.Checkpoint recoveryCheckpoint,protected java.util.concurrent.atomic.AtomicLong uriCount