class SocketFactoryWithTimeout extends SocketFactory {
        protected int connectTimeoutMs = 0;
        
        public int getConnectTimeoutMs() {
            return connectTimeoutMs;
        }

        public void setConnectTimeoutMs(int connectTimeoutMs) {
            this.connectTimeoutMs = connectTimeoutMs;
        }

        public Socket createSocket() {
            return new Socket();
        }

        public Socket createSocket(String host, int port) throws IOException,
                UnknownHostException {
            Socket sock = createSocket();
            sock.connect(new InetSocketAddress(host, port), connectTimeoutMs);
            return sock;
        }

        public Socket createSocket(InetAddress host, int port)
                throws IOException {
            Socket sock = createSocket();
            sock.connect(new InetSocketAddress(host, port), connectTimeoutMs);
            return sock;
        }

        public Socket createSocket(String host, int port,
                InetAddress localHost, int localPort) throws IOException,
                UnknownHostException {
            Socket sock = createSocket();
            sock.bind(new InetSocketAddress(localHost, localPort));
            sock.connect(new InetSocketAddress(host, port), connectTimeoutMs);
            return sock;
        }

        public Socket createSocket(InetAddress address, int port,
                InetAddress localAddress, int localPort) throws IOException {
            Socket sock = createSocket();
            sock.bind(new InetSocketAddress(localAddress, localPort));
            sock.connect(new InetSocketAddress(address, port), connectTimeoutMs);
            return sock;
        }         
        
    }

    protected SocketFactoryWithTimeout socketFactory;

    /**
     * Fetches a document from an FTP server.
     * 
     * @param curi      the URI of the document to fetch
     * @param client    the FTPClient to use for the fetch
     * @param recorder  the recorder to preserve the document in
     * @throws IOException  if a network or protocol error occurs
     * @throws InterruptedException  if the thread is interrupted
     */
    private void fetch(CrawlURI curi, ClientFTP client, Recorder recorder) 
    throws IOException, InterruptedException {<FILL_FUNCTION_BODY>

        // Connect to the FTP server.
        UURI uuri = curi.getUURI();
        int port = uuri.getPort();
        if (port == -1) {
            port = 21;
        }

        if (socketFactory == null) {
            socketFactory = new SocketFactoryWithTimeout();
        }
        socketFactory.setConnectTimeoutMs(getSoTimeoutMs());
        client.setSocketFactory(socketFactory);
        client.setConnectTimeout(getSoTimeoutMs());
        client.setDefaultTimeout(getSoTimeoutMs());
        client.setDataTimeout(getSoTimeoutMs());
        
        client.connect(uuri.getHost(), port);
        
        client.setSoTimeout(getSoTimeoutMs());  // must be after connect()
        
        // Authenticate.
        String[] auth = getAuth(curi);
        client.login(auth[0], auth[1]);
        
        // The given resource may or may not be a directory.
        // To figure out which is which, execute a CD command to
        // the UURI's path.  If CD works, it's a directory.
        boolean isDirectory = client.changeWorkingDirectory(uuri.getPath());

        // Get a data socket.  This will either be the result of a NLST
        // command for a directory, or a RETR command for a file.
        int command;
        String path;
        if (isDirectory) {
            curi.getAnnotations().add("ftpDirectoryList");
            command = FTPCommand.NLST;
            client.setFileType(FTP.ASCII_FILE_TYPE);
            path = ".";
        } else { 
            command = FTPCommand.RETR;
            client.setFileType(FTP.BINARY_FILE_TYPE);
            path = uuri.getPath();
        }

        client.enterLocalPassiveMode();
        Socket socket = null;

        try {
            socket = client.openDataConnection(command, path);

            // if "227 Entering Passive Mode" these will get reset later
            curi.setFetchStatus(client.getReplyCode());
            curi.getData().put(A_FTP_FETCH_STATUS, client.getReplyStrings()[0]);
        } catch (IOException e) {
            // try it again, see AbstractFrontier.needsRetrying()
            curi.setFetchStatus(FetchStatusCodes.S_CONNECT_LOST);
        }

        // Save the streams in the CURI, where downstream processors
        // expect to find them.
        if (socket != null) {
            if (socket.getSoTimeout() != getSoTimeoutMs()) {
                logger.warning("data socket timeout " + socket.getSoTimeout() + "ms is not expected value " + getSoTimeoutMs() + "ms");
            }
            // Shall we get a digest on the content downloaded?
            boolean digestContent = getDigestContent();
            String algorithm = null; 
            if (digestContent) {
                algorithm = getDigestAlgorithm();
                recorder.getRecordedInput().setDigest(algorithm);
                recorder.getRecordedInput().startDigest();
            } else {
                // clear
                recorder.getRecordedInput().setDigest((MessageDigest)null);
            }

            curi.setServerIP(socket.getInetAddress().getHostAddress());
                    
            try {
                saveToRecorder(curi, socket, recorder);
            } finally {
                recorder.close();
                client.closeDataConnection(); // does socket.close()
                curi.setContentSize(recorder.getRecordedInput().getSize());

                // "226 Transfer complete."
                client.getReply();
                curi.setFetchStatus(client.getReplyCode());
                curi.getData().put(A_FTP_FETCH_STATUS, client.getReplyStrings()[0]);
                
                if (isDirectory) {
                    curi.setContentType("text/plain");
                } else {
                    curi.setContentType("application/octet-stream");
                }
                
                if (logger.isLoggable(Level.FINE)) {
                    logger.fine("read " + recorder.getRecordedInput().getSize()
                            + " bytes from ftp data socket");
                }

                if (digestContent) {
                    curi.setContentDigest(algorithm,
                        recorder.getRecordedInput().getDigestValue());
                }
            }

            if (isDirectory) {
                extract(curi, recorder);
            }
        } else {
            // no data - without this, content size is -1
            curi.setContentSize(0);
        }

        addParent(curi);
    
<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