class ARCWriterProcessor extends WriterPoolProcessor {

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

    private static final Logger logger = 
        Logger.getLogger(ARCWriterProcessor.class.getName());

    public long getDefaultMaxFileSize() {
        return 100000000L; // 100 SI mega-bytes (10^8 bytes)
    }
    public List<ConfigPath> getDefaultStorePaths() {
        List<ConfigPath> paths = new ArrayList<ConfigPath>();
        paths.add(new ConfigPath("arcs default store path", "arcs"));
        return paths;
    }

    private transient List<String> cachedMetadata;

    public ARCWriterProcessor() {
    }

    @Override
    protected void setupPool(AtomicInteger serialNo) {
        setPool(new ARCWriterPool(serialNo, this, getPoolMaxActive(), getMaxWaitForIdleMs()));
    }

    /**
     * Writes a CrawlURI and its associated data to store file.
     *
     * Currently this method understands the following uri types: dns, http, 
     * and https.
     *
     * @param curi CrawlURI to process.
     */
    protected ProcessResult innerProcessResult(CrawlURI curi) {<FILL_FUNCTION_BODY>}
    
    protected ProcessResult write(CrawlURI curi, long recordLength, 
            InputStream in, String ip)
    throws IOException {
        WriterPoolMember writer = getPool().borrowFile();
        long position = writer.getPosition();
        // See if we need to open a new file because we've exceeded maxBytes.
        // Call to checkFileSize will open new file if we're at maximum for
        // current file.
        writer.checkSize();
        if (writer.getPosition() != position) {
            // We just closed the file because it was larger than maxBytes.
            // Add to the totalBytesWritten the size of the first record
            // in the file, if any.
            addTotalBytesWritten(writer.getPosition() - position);
            position = writer.getPosition();
        }
        
        ARCWriter w = (ARCWriter)writer;
        try {
            if (in instanceof ReplayInputStream) {
                w.write(curi.toString(), curi.getContentType(),
                    ip, curi.getFetchBeginTime(),
                    recordLength, (ReplayInputStream)in);
            } else {
                w.write(curi.toString(), curi.getContentType(),
                    ip, curi.getFetchBeginTime(),
                    recordLength, in);
            }
        } catch (IOException e) {
            // Invalidate this file (It gets a '.invalid' suffix).
            getPool().invalidateFile(writer);
            // Set the writer to null otherwise the pool accounting
            // of how many active writers gets skewed if we subsequently
            // do a returnWriter call on this object in the finally block.
            writer = null;
            throw e;
        } finally {
            if (writer != null) {
            	addTotalBytesWritten(writer.getPosition() - position);
                getPool().returnFile(writer);
                
                String filename = writer.getFile().getName();
                if (filename.endsWith(ArchiveFileConstants.OCCUPIED_SUFFIX)) {
                    filename = filename.substring(0, filename.length() - ArchiveFileConstants.OCCUPIED_SUFFIX.length());
                }
                curi.addExtraInfo("arcFilename", filename);
                
                Map<String,Object>[] history = curi.getFetchHistory();
                if (history != null && history[0] != null) {
                    history[0].put(A_WRITE_TAG, filename);
                }
            }
        }
        return checkBytesWritten();
    }

    public List<String> getMetadata() {
        if (METADATA_TEMPLATE == null) {
            return null;
        }
        
        if (cachedMetadata != null) {
            return cachedMetadata;
        }
                
        String meta = METADATA_TEMPLATE;
        meta = replace(meta, "${VERSION}", ArchiveUtils.VERSION);
        meta = replace(meta, "${HOST}", getHostName());
        meta = replace(meta, "${IP}", getHostAddress());
        
        if (meta != null) {
            meta = replace(meta, "${JOB_NAME}", getMetadataProvider().getJobName());
            meta = replace(meta, "${DESCRIPTION}", getMetadataProvider().getDescription());
            meta = replace(meta, "${OPERATOR}", getMetadataProvider().getOperator());
            // TODO: fix this to match job-start-date (from UI or operator setting)
            // in the meantime, don't include a slightly-off date
            // meta = replace(meta, "${DATE}", GMT());
            meta = replace(meta, "${USER_AGENT}", getMetadataProvider().getUserAgent());
            meta = replace(meta, "${FROM}", getMetadataProvider().getOperatorFrom());
            meta = replace(meta, "${ROBOTS}", getMetadataProvider().getRobotsPolicyName());
        }

        this.cachedMetadata = Collections.singletonList(meta);
        return this.cachedMetadata;
        // ${VERSION}
        // ${HOST}
        // ${IP}
        // ${JOB_NAME}
        // ${DESCRIPTION}
        // ${OPERATOR}
        // ${DATE}
        // ${USER_AGENT}
        // ${FROM}
        // ${ROBOTS}

    }

    private static String replace(String meta, String find, String replace) {
        replace = StringUtils.defaultString(replace);
        replace = StringEscapeUtils.escapeXml(replace);
        return meta.replace(find, replace);
    }
    
    private static String getHostName() {
        try {
            return InetAddress.getLocalHost().getCanonicalHostName();
        } catch (UnknownHostException e) {
            logger.log(Level.SEVERE, "Could not get local host name.", e);
            return "localhost";
        }
    }
    
    private static String getHostAddress() {
        try {
            return InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            logger.log(Level.SEVERE, "Could not get local host address.", e);
            return "localhost";
        }        
    }

    private static String readMetadataTemplate() {
        InputStream input = ARCWriterProcessor.class.getResourceAsStream(
                "arc_metadata_template.xml");
        if (input == null) {
            logger.severe("No metadata template.");
            return null;
        }
        try {
            return IOUtils.toString(input);
        } catch (IOException e) {
            throw new IllegalStateException(e);
        } finally {
            IOUtils.closeQuietly(input);
        }
    }
}

        long recordLength = getRecordedSize(curi);
        
        ReplayInputStream ris = null;
        try {
            if (shouldWrite(curi)) {
                ris = curi.getRecorder().getRecordedInput()
                        .getReplayInputStream();
                return write(curi, recordLength, ris, getHostAddress(curi));
            } else {
                logger.info("does not write " + curi.toString());
                copyForwardWriteTagIfDupe(curi);
            }
         } catch (IOException e) {
            curi.getNonFatalFailures().add(e);
            logger.log(Level.SEVERE, "Failed write of Record: " +
                curi.toString(), e);
        } finally {
            IOUtils.closeQuietly(ris);
        }
        return ProcessResult.PROCEED;
    
<methods>public void <init>() ,public List<java.io.File> calcOutputDirs() ,public void doCheckpoint(org.archive.checkpointing.Checkpoint) throws java.io.IOException,public boolean getCompress() ,public org.archive.spring.ConfigPath getDirectory() ,public boolean getFrequentFlushes() ,public long getMaxFileSizeBytes() ,public long getMaxTotalBytesToWrite() ,public int getMaxWaitForIdleMs() ,public abstract List<java.lang.String> getMetadata() ,public org.archive.modules.CrawlMetadata getMetadataProvider() ,public int getPoolMaxActive() ,public java.lang.String getPrefix() ,public org.archive.modules.net.ServerCache getServerCache() ,public boolean getSkipIdenticalDigests() ,public boolean getStartNewFilesOnCheckpoint() ,public List<org.archive.spring.ConfigPath> getStorePaths() ,public java.lang.String getTemplate() ,public int getWriteBufferSize() ,public void setCompress(boolean) ,public void setDirectory(org.archive.spring.ConfigPath) ,public void setFrequentFlushes(boolean) ,public void setMaxFileSizeBytes(long) ,public void setMaxTotalBytesToWrite(long) ,public void setMaxWaitForIdleMs(int) ,public void setMetadataProvider(org.archive.modules.CrawlMetadata) ,public void setPoolMaxActive(int) ,public void setPrefix(java.lang.String) ,public void setServerCache(org.archive.modules.net.ServerCache) ,public void setSkipIdenticalDigests(boolean) ,public void setStartNewFilesOnCheckpoint(boolean) ,public void setStorePaths(List<org.archive.spring.ConfigPath>) ,public void setTemplate(java.lang.String) ,public void setWriteBufferSize(int) ,public synchronized void start() ,public void stop() <variables>protected static final java.lang.String ANNOTATION_UNWRITTEN,protected boolean compress,protected org.archive.spring.ConfigPath directory,protected boolean frequentFlushes,private static final java.util.logging.Logger logger,protected long maxFileSizeBytes,protected long maxTotalBytesToWrite,protected int maxWaitForIdleMs,private transient WriterPool pool,protected int poolMaxActive,protected java.lang.String prefix,private java.util.concurrent.atomic.AtomicInteger serial,private static final long serialVersionUID,protected transient org.archive.modules.net.ServerCache serverCache,protected boolean skipIdenticalDigests,protected boolean startNewFilesOnCheckpoint,protected List<org.archive.spring.ConfigPath> storePaths,protected java.lang.String template,private java.util.concurrent.atomic.AtomicLong totalBytesWritten,protected int writeBufferSize