
package com.limegroup.gnutella.downloader;

import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Set;

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

import com.limegroup.gnutella.Assert;
import com.limegroup.gnutella.AssertFailure;
import com.limegroup.gnutella.InsufficientDataException;
import com.limegroup.gnutella.RemoteFileDesc;
import com.limegroup.gnutella.RouterService;
import com.limegroup.gnutella.altlocs.AlternateLocation;
import com.limegroup.gnutella.http.ProblemReadingHeaderException;
import com.limegroup.gnutella.io.IOStateObserver;
import com.limegroup.gnutella.io.NIODispatcher;
import com.limegroup.gnutella.settings.DownloadSettings;
import com.limegroup.gnutella.statistics.DownloadStat;
import com.limegroup.gnutella.statistics.NumericalDownloadStat;
import com.limegroup.gnutella.tigertree.HashTree;
import com.limegroup.gnutella.util.IOUtils;
import com.limegroup.gnutella.util.IntervalSet;
import com.limegroup.gnutella.util.Sockets;

/**
 * Class that performs the logic of downloading a file from a single host.
 */
public class DownloadWorker {
    /*
      A worker follows these steps:
      
      CONNECTING:
         Establish a TCP connection to the host in the RFD.
         If unable to connect, exit.
         If able to connect, continue processing the download.
         
         This step is characterized by following:
          establishConnection -> [push|direct] -> startDownload
         
      DOWNLOADING:
         The download enters a state machine, which loops forever until
         either an error occurs or the download is finished.         
         The flow is similar to:
         
         while(true) {
             if(can request thex) {
                 request and download thex             
             if(needs to consume body)
                 consume body             
             assign and request             
             if(ready to download)
                 do download
             else if(queued)
                 wait queue time
             else
                 exit
         }
         
         except it is performed asynchronously via a state machine.
         
         The states are entered via httpLoop and progress through
         calls of incrementState(ConnectionStatus).  It moves through
         the following steps:                
            - requestThexIfNeeded
            - downloadThexIfNeeded
            - consumeBodyIfNeeded
            - assignAndRequest
                - assignWhite or assignGrey
                - completeAssignAndRequest
                - completeAssignWhite or completeAssignGrey
            - httpRequestFinished
                - beginDownload, handleQueued, or finishHttpLoop
                
            Each 'if needed' method can return true or false.  True means that
            an operation is being performend and upon success or failure the
            state machine will continue.  Success generally calls incrementState
            again to move to the next state.  Failure generally calls finishHttpLoop
            to stop the download.  False means the operation does not need to be
            performed and the next state can be immediately processed.
            
            The assignAndRequest step has two parts:
            a.  Grab a part of the file to download. If there is unclaimed area on
                the file grab that, otherwise try to steal claimed area from another 
                worker
            b.  Send http headers to the uploader on the tcp connection 
                established in step 1. The uploader may or may not be able to 
                upload at this time. If the uploader can't upload, it's 
                important that the leased area be restored to the state 
                they were in before we started trying. However, if the http 
                handshaking was successful, the downloader can keep the 
                part it obtained.

            Both assignWhite & assignGrey will schedule the HTTP request (part b above)
            and continue afterwards by calling completeAssignAndRequest even if the
            request had an exception.  This is done so that any read headers can be
            parsed and accounted for, such as alternate locations.
    
    PUSH DOWNLOADS NOTE:
      For push downloads, the acceptDownload(file, Socket, index, clientGUI) 
      method of ManagedDownloader is called from DownloadManager.
      This method needs to notify the appropriate downloader so that it can use
      the socket. 
      
      When establishConnection() realizes that it needs to do a push, it gives the
      manager its HTTPConnectObserver (a ConnectObserver) and a mini-RFD.  When the manager
      is notified that a push was accepted (via acceptDownload) with that mini-RFD,
      it will notify the HTTPConnectObserver using handleConnect(Socket).
      
      Note: The establishConnection method schedules a Runnable to remove the observer
      in a short amount of time (about 9 seconds).  If the observer hasn't already 
      connected, it assumes the push failed and terminates by calling shutdown().
      
      If the push was done by a multicast RFD, a failure to connect will proceed to trying
      a direct connection.  Otherwise (the push was done because no direct connect was
      possible, or because a direct connect failed), the failure of a push means that the
      download cannot proceed.
      
     CONNECTION ESTABLISHMENT NOTE:
       All connection establish, push or direct, is done via callbacks.  There is no thread
       blocking on connection establishment.  When a connection either succeeds a ConnectObserver's
       handleConnect(Socket) is called, which will ultimately attempt to start the download via
       startDownload.  If the connection attempt failed, the ConnectObserver's shutdown method
       is called and no thread is ever created.
    */
    private static final Log LOG = LogFactory.getLog(DownloadWorker.class);
    
    ///////////////////////// Policy Controls ///////////////////////////
    /** The smallest interval that can be split for parallel download */
    private static final int MIN_SPLIT_SIZE=16*1024;      //16 KB
    
    /** The lowest (cumulative) bandwith we will accept without stealing the
     * entire grey area from a downloader for a new one */
    private static final float MIN_ACCEPTABLE_SPEED = 
		DownloadSettings.MAX_DOWNLOAD_BYTES_PER_SEC.getValue() < 8 ? 
		0.1f:
		0.5f;
    
    /** 
     * The speed of download workers that haven't been started yet or do not 
     * have enough measurements
     */
    private static final int UNKNOWN_SPEED = -1;
    
    /** The time to wait trying to establish each normal connection, in
     *  milliseconds.*/
    private static final int NORMAL_CONNECT_TIME=10000; //10 seconds
    /** The time to wait trying to establish each push connection, in
     *  milliseconds.  This needs to be larger than the normal time. */
    private static final int PUSH_CONNECT_TIME=20000;  //20 seconds
    /** The time to wait trying to establish a push connection
     * if only a UDP push has been sent (as is in the case of altlocs) */
    private static final int UDP_PUSH_CONNECT_TIME=6000; //6 seconds
    
    /**
     * The number of seconds to wait for hosts that don't have any ranges we
     *  would be interested in.
     */
    private static final int NO_RANGES_RETRY_AFTER = 60 * 5; // 5 minutes
    
    /**
     * The number of seconds to wait for hosts that failed once.
     */
    private static final int FAILED_RETRY_AFTER = 60 * 1; // 1 minute
    
    /**
     * The number of seconds to wait for a busy host (if it didn't give us a
     * retry after header) if we don't have any active downloaders.
     *
     * Note that there are some acceptable problems with the way this 
     * values are used.  Namely, if we have sources X & Y and source
     * X is tried first, but is busy, its busy-time will be set to
     * 1 minute.  Then source Y is tried and is accepted, source X
     * will still retry after 1 minute.  This 'problem' is considered
     * an acceptable issue, given the complexity of implementing
     * a method that will work under the circumstances.
     */
    public static final int RETRY_AFTER_NONE_ACTIVE = 60 * 1; // 1 minute
    
    /**
     * The minimum number of seconds to wait for a busy host if we do
     * have some active downloaders.
     *
     * Note that there are some acceptable problems with the way this
     * values are used.  Namely, if we have sources X & Y and source
     * X is tried first and is accepted.  Then source Y is tried and
     * is busy, so its busy-time is set to 10 minutes.  Then X disconnects,
     * leaving Y with 9 or so minutes left before being retried, despite
     * no other sources available.  This 'problem' is considered
     * an acceptable issue, given the complexity of implementing
     * a method that will work under the circumstances.
     */
    private static final int RETRY_AFTER_SOME_ACTIVE = 60 * 10; // 10 minutes

    private final ManagedDownloader _manager;
    private final RemoteFileDesc _rfd;
    private final VerifyingFile _commonOutFile;
    
    /**
     * Whether I was interrupted before starting
     */
    private volatile boolean _interrupted;
    
    /**
     * The downloader that will do the actual downloading
     * TODO: un-volatilize after fixing the assertion failures
     */
    private volatile HTTPDownloader _downloader;
    
    /**
     * Whether I should release the ranges that I have leased for download
     * TODO: un-volatilize after fixing the assertion failures
     */
    private volatile boolean _shouldRelease;
    
    /**
     * The name this worker has in toString & threads.
     */
    private final String _workerName;
    
    /** The observer used for direct connection establishment. */
    private DirectConnector _connectObserver;
    
    /** The current state of the non-blocking download. */
    private DownloadState _currentState;
    
    /**
     *  Whether or not the worker is involved in a stealing operation
     *  (as either a thief or victim).
     */
    private volatile boolean _stealing;
    
    DownloadWorker(ManagedDownloader manager, RemoteFileDesc rfd, VerifyingFile vf) {
        _manager = manager;
        _rfd = rfd;
        _commonOutFile = vf;
        _currentState = new DownloadState();
        
        // if we'll be debugging, we want to distinguish the different workers
        if (LOG.isDebugEnabled()) {
            _workerName = "DownloadWorker for " + _manager.getSaveFile().getName() + " #" + System.identityHashCode(this);
        } else {
            _workerName = "DownloaderWorker";
        }        
    }
    
    /**
     * Starts this DownloadWorker's connection establishment.
     */
    public void start() {
        if(LOG.isDebugEnabled())
            LOG.debug("Starting worker: " + _workerName);
        establishConnection();
    }
    
    /**
     * Initializes the HTTPDownloader with whatever AltLocs we have discovered so far.
     * These will be cleared out after the first write.  From then on, only newly successful
     * RFDS will be sent as Alts.
     */
    private void initializeAlternateLocations() {
        int count = 0;
        for(Iterator iter = _manager.getValidAlts().iterator(); 
          iter.hasNext() && count < 10; count++) {
            AlternateLocation current = (AlternateLocation)iter.next();
            _downloader.addSuccessfulAltLoc(current);
        }
        
        count = 0;
        for(Iterator iter = _manager.getInvalidAlts().iterator(); 
          iter.hasNext() && count < 10; count++) {
            AlternateLocation current = (AlternateLocation)iter.next();
            _downloader.addFailedAltLoc(current);
        }
    }
    
    /**
     * Begins the state machine for processing this download.
     */
    private void httpLoop() {
        LOG.debug("Starting HTTP Loop");
        incrementState(null);
    }
    
    /**
     * Notification that a state has finished.
     * This kicks off the next stage if necessary.
     */
    public void incrementState(ConnectionStatus status) {
        if(LOG.isTraceEnabled())
            LOG.trace("WORKER: " + this + ", State Changed, Current: " + _currentState + ", status: " + status);
        
        if(_interrupted) {
            finishHttpLoop();
            return;
        }
        
        switch(_currentState.getCurrentState()) {
        case DownloadState.DOWNLOADING:
            releaseRanges();
        case DownloadState.QUEUED:
        case DownloadState.BEGIN:
            _currentState.setHttp11(_rfd.isHTTP11());
            _currentState.setState(DownloadState.REQUESTING_THEX);
            if(requestTHEXIfNeeded())
                break; // wait for callback
        
        case DownloadState.REQUESTING_THEX:
            _currentState.setState(DownloadState.DOWNLOADING_THEX);
            if(downloadThexIfNeeded())
                break;
        
        case DownloadState.DOWNLOADING_THEX:
            _currentState.setState(DownloadState.CONSUMING_BODY);
            if(consumeBodyIfNeeded())
                break; // wait for callback
            
        case DownloadState.CONSUMING_BODY:
            _downloader.forgetRanges();
            if(status == null || !status.isQueued()) {
                _currentState.setState(DownloadState.REQUESTING_HTTP);
                if(!assignAndRequest()) { // no data
                    finishHttpLoop();
                }
                break; // wait for callback (or exit)
            }
            
        case DownloadState.REQUESTING_HTTP:
            httpRequestFinished(status);
            break;
        default:
            throw new IllegalStateException("bad state: " + _currentState);
        }
    }
    
    /**
     * Consumes the body of an HTTP Request if necessary.
     * If consumption is needed, this will return true and schedule
     * a callback to continue.  Otherwise it will return false.
     * 
     * @return true if the body is scheduled for consumption, false if processing should continue.
     */
    private boolean consumeBodyIfNeeded() {
        if(_downloader.isBodyConsumed()) {
            LOG.debug("Not consuming body.");
            return false;
        }
        
        _downloader.consumeBody(new State() {
            protected void handleState(boolean success) {
                if(!success)
                    handleRFDFailure(_rfd);
            }
        });
        return true;
    }
    
    /**
     * Handles a failure of an RFD.
     * @param rfd
     */
    private void handleRFDFailure(RemoteFileDesc rfd) {
        _rfd.incrementFailedCount();
        
        // if this RFD had a failure, try it again.
        if( _rfd.getFailedCount() < 2 ) {
            //set retry after, wait a little before retrying this RFD
            _rfd.setRetryAfter(FAILED_RETRY_AFTER);
            _manager.addRFD(_rfd);
        } else //tried the location twice -- it really is bad
            _manager.informMesh(_rfd, false);
    }
    
    /**
     * Notification that assign&Request has finished.
     * This will:
     *   - Finish the download if no file was available.
     *   - Loop again if the requested range was unavailable but other data is available.
     *   - Queue up if we were instructed to be queued.
     *   - Download if we were told to download.
     *   
     *  In all events, either the download completely finishes or a callback
     *  is eventually notified of success or failure & the state continues
     *  moving.
     *  
     * @param status
     */
    private void httpRequestFinished(ConnectionStatus status) {
        if(LOG.isDebugEnabled())
            LOG.debug("HTTP req finished, status: " + status);

        _manager.addPossibleSources(_downloader.getLocationsReceived());
        
        if(status.isNoData() || status.isNoFile()) {
            finishHttpLoop();
        } else {
            if(!status.isConnected())
                releaseRanges();
            
            // After A&R, we got a non queued response.
            if(!status.isQueued())
                _manager.removeQueuedWorker(this);
            
            if(status.isPartialData()) {
                _currentState.setState(DownloadState.BEGIN);
                incrementState(null);
            } else {
                Assert.that( status.isQueued() || status.isConnected() );
                boolean queued = _manager.killQueuedIfNecessary(this, !status.isQueued() ? -1 : status.getQueuePosition());
                
                if(status.isConnected()) {
                    _currentState.setState(DownloadState.DOWNLOADING);
                    beginDownload();
                } else if (!queued) { // If we were told not to queue.
                    finishHttpLoop();
                } else {
                    handleQueued(status);
                }
            }
        }
    }
    
    /**
     * Begins the process of downloading.  When downloading finishes,
     * this will either finish the download (if an error occurred) or
     * move to the next state.
     * 
     * A succesful download will reset the failed count on the RFD.
     * A DiskException while downloading will notify the manager of
     * a problem.
     */
    private void beginDownload() {
        try {
            _downloader.doDownload(new State() {
                protected void handleState(boolean success) {
                    if(success) {
                        _rfd.resetFailedCount();
                        if(_currentState.isHttp11())
                            DownloadStat.SUCCESSFUL_HTTP11.incrementStat();
                        else
                            DownloadStat.SUCCESSFUL_HTTP10.incrementStat();
                    } else {
                        if(_currentState.isHttp11())
                            DownloadStat.FAILED_HTTP11.incrementStat();
                        else
                            DownloadStat.FAILED_HTTP10.incrementStat();
                        _manager.workerFailed(DownloadWorker.this);
                    }
                    
                    // if we got too corrupted, notify the user
                    if (_commonOutFile.isHopeless())
                        _manager.promptAboutCorruptDownload();
    
                    int stop = _downloader.getInitialReadingPoint() + _downloader.getAmountRead();
                    if (LOG.isDebugEnabled())
                        LOG.debug("WORKER: terminating from " + _downloader + " at " + stop + " error? " + !success);
                    
                    synchronized (_manager) {
                        if (!success) {
                            _downloader.stop();
                            handleRFDFailure(_rfd);
                        } else {
                            _manager.informMesh(_rfd, true);
                            if (!_currentState.isHttp11()) // no need to add http11 _activeWorkers to files
                                _manager.addRFD(_rfd);
                        }
                    }
                }
            });
        } catch(SocketException se) {
            finishHttpLoop();
        }
    }
    
    /**
     * Determines if we should request a tiger tree from the remote computer.
     * This will return true if a request is going to be performed and false
     * otherwise.  If this returns true, a callback will eventually increment
     * the state or finish the download completely.
     * 
     * @return true if the request is scheduled to be sent, false if processing should continue.
     */
    private boolean requestTHEXIfNeeded() {
        boolean shouldRequest = false;
        synchronized (_commonOutFile) {
            if (!_commonOutFile.isHashTreeRequested()) {
                HashTree ourTree = _commonOutFile.getHashTree();
    
                // request THEX from the _downloader if (the tree we have
                // isn't good enough or we don't have a tree) and another
                // worker isn't currently requesting one
                shouldRequest = _downloader.hasHashTree()
                             && _manager.getSHA1Urn() != null
                             && (ourTree == null || !ourTree.isDepthGoodEnough());
                    
                if (shouldRequest)
                    _commonOutFile.setHashTreeRequested(true);
            }
        }

        if (shouldRequest) {
            _downloader.requestHashTree(_manager.getSHA1Urn(), new State() {
                protected void handleState(boolean success) {}
            });
        }
        
        return shouldRequest;
    }
    
    /**
     * Begins a THEX download if it was just requested.
     * 
     * If the request failed, this will immediately increment the state
     * so that the body of the response can be consumed.  Otherwise
     * it will schedule a download to take place and increment
     * the state when finished.
     * 
     * @return true if the download was scheduled, false if processing should continue.
     */
    private boolean downloadThexIfNeeded() {
        if(!_downloader.isRequestingThex())
            return false;
        
        ConnectionStatus status = _downloader.parseThexResponseHeaders();
        if(!status.isConnected()) {
            // retry this RFD without THEX, since that's why it failed.
            _rfd.setTHEXFailed();
            incrementState(status);
        } else {
            _manager.removeQueuedWorker(this);
            _downloader.downloadThexBody(_manager.getSHA1Urn(), new State() {
                protected void handleState(boolean success) {
                    synchronized(_commonOutFile) {
                        _commonOutFile.setHashTreeRequested(false);
                        HashTree newTree = _downloader.getHashTree();
                        if(LOG.isDebugEnabled())
                            LOG.debug("Downloaded tree: " + newTree);
                        if(newTree != null) {
                            HashTree oldTree = _commonOutFile.getHashTree();
                            if(newTree.isBetterTree(oldTree))
                                _commonOutFile.setHashTree(newTree);
                        }
                    }
                }
            });
        }
        
        return true;
    }
    
    /**
     * Release the ranges assigned to our downloader  
     */
    private void releaseRanges() {
        
        if (!_shouldRelease)
            return;
        _shouldRelease = false;
        
        // do not release if the file is complete
        if (_commonOutFile.isComplete())
            return;
        
        HTTPDownloader downloader = _downloader;
        int high, low;
        synchronized(downloader) {
        	
            // If this downloader was a thief and had to skip any ranges, do not
            // release them.
            low = downloader.getInitialReadingPoint() + downloader.getAmountRead();
            low = Math.max(low,downloader.getInitialWritingPoint());
            high = downloader.getInitialReadingPoint() + downloader.getAmountToRead()-1;
        }
        
        if( (high-low)>=0) {//dloader failed to download a part assigned to it?
            
            if (LOG.isDebugEnabled())
                LOG.debug("releasing ranges "+new Interval(low,high));

            try {
            	_commonOutFile.releaseBlock(new Interval(low,high));
            } catch (AssertFailure bad) {
            	downloader.createAssertionReport(bad);
            }
            
            downloader.forgetRanges();
        } else 
			LOG.debug("nothing to release!");
    }
    
    /**
     * Schedules a callback for a queued worker.
     *
     * @return true if we need to tell the manager to churn another
     *         connection and let this one die, false if we are
     *         going to try this connection again.
     */
    private void handleQueued(ConnectionStatus status) {
        // make sure that we're not in _downloaders if we're
        // sleeping/queued. this would ONLY be possible
        // if some uploader was misbehaved and queued
        // us after we succesfully managed to download some
        // information. despite the rarity of the situation,
        // we should be prepared.
        _manager.removeActiveWorker(this);
        
        synchronized(_currentState) {
            if(_interrupted) {
                LOG.debug("Exiting from queueing");
                return;
            }
            
            LOG.debug("Queueing");
            _currentState.setState(DownloadState.QUEUED);
        }
        
        RouterService.schedule(new Runnable() {
            public void run() {
                LOG.debug("Queue time up");
                
                synchronized(_currentState) {
                    if(_interrupted) {
                        LOG.warn("WORKER: interrupted while waiting in queue " + _downloader);
                        return;
                    }
                }
                
                NIODispatcher.instance().invokeLater(new Runnable() {
                    public void run() {
                        incrementState(null);
                    }
                });
            }
        }, status.getQueuePollTime(), 0);
    }
    
    /** 
     * Attempts to establish a connection to the host in RFD.
     * 
     * This will return immediately, scheduling callbacks for the connection
     * events.  The appropriate ConnectObserver (Push or Direct) will be
     * notified via handleConnect if succesful or shutdown if not.  From there,
     * the rest of the download may start. 
     */
    private void establishConnection() {
        if(LOG.isTraceEnabled())
            LOG.trace("establishConnection(" + _rfd + ")");
        
        // this rfd may still be useful remember it
        if (_manager.isCancelled() || _manager.isPaused() || _interrupted) {
            _manager.addRFD(_rfd);
            finishWorker();
            return;
        }

        final boolean needsPush = _rfd.needsPush();
        
        synchronized (_manager) {
            int state = _manager.getState();
            //If we're just increasing parallelism, stay in DOWNLOADING
            //state.  Otherwise the following call is needed to restart
            //the timer.
            if (_manager.getNumDownloaders() == 0 && state != ManagedDownloader.COMPLETE && 
                state != ManagedDownloader.ABORTED && state != ManagedDownloader.GAVE_UP && 
                state != ManagedDownloader.DISK_PROBLEM && state != ManagedDownloader.CORRUPT_FILE && 
                state != ManagedDownloader.HASHING && state != ManagedDownloader.SAVING) {
                    if(_interrupted)
                        return; // we were signalled to stop.
                    _manager.setState(ManagedDownloader.CONNECTING);
                }
        }

        if (LOG.isDebugEnabled())   
            LOG.debug("WORKER: attempting connect to " + _rfd.getHost() + ":" + _rfd.getPort());        
        
        DownloadStat.CONNECTION_ATTEMPTS.incrementStat();

        if (_rfd.isReplyToMulticast()) {
            // Start with a push connect, fallback to a direct connect, and do
            // not forget the RFD upon push failure.
            connectWithPush(new PushConnector(false, true));
        } else if (!needsPush) {
            // Start with a direct connect, fallback to a push connect.
            connectDirectly(new DirectConnector(true));
        } else {
            // Start with a push connect, do not fallback to a direct connect, and do
            // forgot the RFD upon push failure.
            connectWithPush(new PushConnector(true, false));
        }
    }

    /**
     * Performs actions necessary after the connection process is finished. This will tell the manager this is a bad RFD
     * if no downloader could be created, and stop the downloader if we were interrupted. Returns true if the download
     * should proceed, false otherwise.
     */
    private boolean finishConnect() {
        // if we didn't connect at all, tell the rest about this rfd
        if (_downloader == null) {
            _manager.informMesh(_rfd, false);
            return false;
        } else if (_interrupted) {
            // if the worker got killed, make sure the downloader is stopped.
            _downloader.stop();
            _downloader = null;
            return false;
        }
        return true;
    }
    
    /**
     * Attempts to asynchronously connect through TCP to the remote end.
     * This will return immediately and the given observer will be notified
     * of success or failure.
     */
    private void connectDirectly(DirectConnector observer) {
        if (!_interrupted) {
            if(LOG.isTraceEnabled())
                LOG.trace("WORKER: attempt asynchronous direct connection to: " + _rfd);
            _connectObserver = observer;
            try {
                Socket socket = Sockets.connect(_rfd.getHost(), _rfd.getPort(), NORMAL_CONNECT_TIME, observer);
                if(!observer.isShutdown())
                    observer.setSocket(socket);
            } catch (IOException iox) {
                observer.shutdown();
            }
        } else {
            finishWorker();
        }
    }
    
    /**
     * Attempts to connect by using a push to the remote end.
     * This method will return immediately and the given observer will
     * be notified of success or failure.
     */
    private void connectWithPush(PushConnector observer) {
        if(!_interrupted) {
            if(LOG.isTraceEnabled())
                LOG.trace("WORKER: attempt push connection to: " + _rfd);
            _connectObserver = null;
            
            //When the push is complete and we have a socket ready to use
            //the acceptor thread is going to notify us using this object
            final PushDetails details = new PushDetails(_rfd.getClientGUID(), _rfd.getHost());
            observer.setPushDetails(details);
            _manager.registerPushObserver(observer, details);
            RouterService.getDownloadManager().sendPush(_rfd, observer);
            RouterService.schedule(new Runnable() {
                public void run() {
                    _manager.unregisterPushObserver(details, true);
                }
            }, _rfd.isFromAlternateLocation() ? UDP_PUSH_CONNECT_TIME : PUSH_CONNECT_TIME, 0);
        } else {
            finishWorker();
        }
    }
    
    String getInfo() {
        if (_downloader != null) {
            synchronized(_downloader) {
                return this + "hashcode " + System.identityHashCode(_downloader) + " will release? "
                + _shouldRelease + " interrupted? " + _interrupted
                + " active? " + _downloader.isActive() 
                + " victim? " + _downloader.isVictim()
                + " initial reading " + _downloader.getInitialReadingPoint()
                + " initial writing " + _downloader.getInitialWritingPoint()
                + " amount to read " + _downloader.getAmountToRead()
                + " amount read " + _downloader.getAmountRead()
                + " is in stealing " + isStealing()+"\n";
            }
        } else 
            return "worker not started";
    }
    
    /** 
     * Assigns a white area or a grey area to a downloader. Sets the state,
     * and checks if this downloader has been interrupted.
     * @param _downloader The downloader to which this method assigns either
     * a grey area or white area.
     * @return the ConnectionStatus.
     */
    private boolean assignAndRequest() {
        if(LOG.isTraceEnabled())
            LOG.trace("assignAndRequest for: " + _rfd);
    
        Interval interval = null;
        try {
            synchronized (_commonOutFile) {
                if (_commonOutFile.hasFreeBlocksToAssign() > 0)
                    interval = pickAvailableInterval();
            }
        } catch (NoSuchRangeException nsre) {
            handleNoRanges();
            return false;
        }
        
        // it is still possible that a worker has died and released their ranges
        // just before we try to steal
        if (interval == null) {
            if(!assignGrey())
                return false;
        } else {
            assignWhite(interval);
        }
        
        return true;
    }
    
    /**
     * Completes the assignAndRequest by incrementing the state using the
     * ConnectionStatus that is generated by processing of the response headers.
     * 
     * @param x Any IOException encountered while processing
     * @param range The range initially requested
     * @param victim The possibly null victim to steal from. 
     */
    private void completeAssignAndRequest(IOException x, Interval range, DownloadWorker victim) {
        ConnectionStatus status = completeAssignAndRequestImpl(x, range, victim);
        if (victim != null) {
            victim.setStealing(false);
            setStealing(false);
        }
        incrementState(status);
    }
    
    /**
     * Completes the assign & request process by parsing the response
     * headers and completing either assignWhite or assignGrey.
     * 
     * If victim is null, it is assumed that we are completing assignGrey.
     * Otherwise, we are completing assignWhite.
     * 
     * @param x Any IOException encountered while processing
     * @param range The range initially requested
     * @param victim The possibly null victim to steal from. 
     * @return
     */
    private ConnectionStatus completeAssignAndRequestImpl(IOException x, Interval range, DownloadWorker victim) {
        try {
            try {
                _downloader.parseHeaders();
            } finally {
                // The IOX passed in here takes priority over
                // any exception from parsing the headers.
                if(x != null)
                    throw x;
            }
            
            if(victim == null)
                completeAssignWhite(range);
            else
                completeAssignGrey(victim, range);
            
        } catch(NoSuchElementException nsex) {
            DownloadStat.NSE_EXCEPTION.incrementStat();
            LOG.debug(_downloader,nsex);
            
            return handleNoMoreDownloaders();
            
        } catch (NoSuchRangeException nsrx) {
            LOG.debug(_downloader,nsrx);

            return handleNoRanges();
            
        } catch(TryAgainLaterException talx) {
            DownloadStat.TAL_EXCEPTION.incrementStat();
            LOG.debug(_downloader,talx);
            
            return handleTryAgainLater();
            
        } catch(RangeNotAvailableException rnae) {
            DownloadStat.RNA_EXCEPTION.incrementStat();
            LOG.debug(_downloader,rnae);
            
            return handleRangeNotAvailable();
            
        } catch (FileNotFoundException fnfx) {
            DownloadStat.FNF_EXCEPTION.incrementStat();
            LOG.debug(_downloader, fnfx);
            
            return handleFileNotFound();
            
        } catch (NotSharingException nsx) {
            DownloadStat.NS_EXCEPTION.incrementStat();
            LOG.debug(_downloader, nsx);
            
            return handleNotSharing();
            
        } catch (QueuedException qx) { 
            DownloadStat.Q_EXCEPTION.incrementStat();
            LOG.debug(_downloader, qx);
            
            return handleQueued(qx.getQueuePosition(),qx.getMinPollTime());
            
        } catch(ProblemReadingHeaderException prhe) {
            DownloadStat.PRH_EXCEPTION.incrementStat();
            LOG.debug(_downloader,prhe);
            
            return handleProblemReadingHeader();
            
        } catch(UnknownCodeException uce) {
            DownloadStat.UNKNOWN_CODE_EXCEPTION.incrementStat();
            LOG.debug(_downloader, uce);
            
            return handleUnknownCode();
            
        } catch (ContentUrnMismatchException cume) {
        	DownloadStat.CONTENT_URN_MISMATCH_EXCEPTION.incrementStat();
            LOG.debug(_downloader, cume);
        	
			return ConnectionStatus.getNoFile();
			
        } catch (IOException iox) {
            DownloadStat.IO_EXCEPTION.incrementStat();
            LOG.debug(_downloader, iox);
            
            return handleIO();
            
        } 
        
        //did not throw exception? OK. we are downloading
        DownloadStat.RESPONSE_OK.incrementStat();
        if(_rfd.getFailedCount() > 0)
            DownloadStat.RETRIED_SUCCESS.incrementStat();    
        
        _rfd.resetFailedCount();

        synchronized(_manager) {
            if (_manager.isCancelled() || _manager.isPaused() || _interrupted) {
                LOG.trace("Stopped in assignAndRequest");
                _manager.addRFD(_rfd);
                return ConnectionStatus.getNoData();
            }
            
            _manager.workerStarted(this);
        }
        
        return ConnectionStatus.getConnected();
    }
    
    /**
     * Schedules a request for the given interval.
     * Upon completion of the request, completeAssignAndRequest will
     * be called with the appropriate parameters.
     */
    private void assignWhite(Interval interval) {
        //Intervals from the IntervalSet set are INCLUSIVE on the high end, but
        //intervals passed to HTTPDownloader are EXCLUSIVE.  Hence the +1 in the
        //code below.  Note connectHTTP can throw several exceptions.
        final int low = interval.low;
        final int high = interval.high; // INCLUSIVE
		_shouldRelease=true;
        _downloader.connectHTTP(low, high + 1, true,_commonOutFile.getBlockSize(), new IOStateObserver() {
            public void handleStatesFinished() {
                completeAssignAndRequest(null, new Interval(low, high), null);
            }

            public void handleIOException(IOException iox) {
                completeAssignAndRequest(iox, null, null);
            }

            public void shutdown() {
                completeAssignAndRequest(new IOException("shutdown"), null, null);
            }
            
        });
    }
    
    /**
     * Completes assigning a white range to a downloader.
     * If the downloader shortened any of the requested ranges,
     * this will release the remaining pieces back to the VerifyingFile.
     * 
     * @param expectedRange
     */
    private void completeAssignWhite(Interval expectedRange) {
        //The _downloader may have told us that we're going to read less data than
        //we expect to read.  We must release the not downloading leased intervals
        //We only want to release a range if the reported subrange
        //was different, and was HIGHER than the low point.
        //in case this worker became a victim during the header exchange, we do not
        //clip any ranges.
        synchronized(_downloader) {
            int low = expectedRange.low;
            int high = expectedRange.high;
            int newLow = _downloader.getInitialReadingPoint();
            int newHigh = (_downloader.getAmountToRead() - 1) + newLow; // INCLUSIVE
            if (newHigh-newLow >= 0) {
                if(newLow > low) {
                    if(LOG.isDebugEnabled())
                        LOG.debug("WORKER:"+
                                " Host gave subrange, different low.  Was: " +
                                low + ", is now: " + newLow);
                    
                    _commonOutFile.releaseBlock(new Interval(low, newLow-1));
                }
                
                if(newHigh < high) {
                    if(LOG.isDebugEnabled())
                        LOG.debug("WORKER:"+
                                " Host gave subrange, different high.  Was: " +
                                high + ", is now: " + newHigh);
                    
                    _commonOutFile.releaseBlock(new Interval(newHigh+1, high));
                }
                
                if(LOG.isDebugEnabled()) {
                    LOG.debug("WORKER:"+
                            " assigning white " + newLow + "-" + newHigh +
                            " to " + _downloader);
                }
            } else 
                LOG.debug("debouched at birth");
        }
    }
    
    /**
     * picks an unclaimed interval from the verifying file
     * 
     * @param http11 whether the downloader is http 11
     * 
     * @throws NoSuchRangeException if the remote host is partial and doesn't 
     * have the ranges we need
     */
    private Interval pickAvailableInterval() throws NoSuchRangeException {
        Interval interval = null;
        //If it's not a partial source, take the first chunk.
        // (If it's HTTP11, take the first chunk up to CHUNK_SIZE)
        if( !_downloader.getRemoteFileDesc().isPartialSource() ) {
            if(_currentState.isHttp11()) {
                interval = _commonOutFile.leaseWhite(findChunkSize());
            } else
                interval = _commonOutFile.leaseWhite();
        }
        
        // If it is a partial source, extract the first needed/available range
        // (If it's HTTP11, take the first chunk up to CHUNK_SIZE)
        else {
            try {
                IntervalSet availableRanges = _downloader.getRemoteFileDesc().getAvailableRanges();

                if (_currentState.isHttp11()) {
                    interval = _commonOutFile.leaseWhite(availableRanges, findChunkSize());
                } else
                    interval = _commonOutFile.leaseWhite(availableRanges);

            } catch (NoSuchElementException nsee) {
                // if nothing satisfied this partial source, don't throw NSEE
                // because that means there's nothing left to download.
                // throw NSRE, which means that this particular source is done.
                throw new NoSuchRangeException();
            }
        }
        
        return interval;
    }

    private int findChunkSize() {
        int chunkSize = _commonOutFile.getChunkSize();
        int free = _commonOutFile.hasFreeBlocksToAssign();
        
        // if we have less than one free chunk, take half of that
        if (free <= chunkSize && _manager.getActiveWorkers().size() > 1) 
            chunkSize = Math.max(MIN_SPLIT_SIZE, free / 2);
        
        return chunkSize;
    }
    
    /**
     * Locates an interval from the slowest downloader and schedules
     * a request with it.  If the current download has partial ranges,
     * there is no slowest download, or the slowest downloader has no
     * ranges available for stealing, this will return false and processing
     * will immediately continue.  Otherwise, this will return true
     * and completeAssignAndRequest will be called when the request completes.
     */
    private boolean assignGrey() {
    	// if I'm currently being stolen from, do not try to steal.
    	// can happen if my thief is exchanging headers.
        if (isStealing())
        	return false;
        
        //If this _downloader is a partial source, don't attempt to steal...
        //too confusing, too many problems, etc...
        if( _downloader.getRemoteFileDesc().isPartialSource() ) {
            handleNoRanges();
            return false;
        }

        final DownloadWorker slowest = findSlowestDownloader();
                        
        if (slowest==null) {//Not using this downloader...but RFD maybe useful
            LOG.debug("didn't find anybody to steal from");
            handleNoMoreDownloaders();
            return false;
        }
		
        // see what ranges is the victim requesting
        final Interval slowestRange = slowest.getDownloadInterval();
        
        if (slowestRange.low == slowestRange.high) {
            handleNoMoreDownloaders();
            return false;
        }
        
        //Note: we are not interested in being queued at this point this
        //line could throw a bunch of exceptions (not queuedException)
        slowest.setStealing(true);
        setStealing(true);
        _downloader.connectHTTP(slowestRange.low, slowestRange.high, false,_commonOutFile.getBlockSize(), new IOStateObserver() {
            public void handleStatesFinished() {
                completeAssignAndRequest(null, slowestRange, slowest);
            }

            public void handleIOException(IOException iox) {
                completeAssignAndRequest(iox, null, slowest);
            }

            public void shutdown() {
                completeAssignAndRequest(new IOException("shutdown"), null, slowest);
            }
            
        });
        
        return true;
    }
    
    /**
     * Completes assigning a grey portion to a downloader.
     * This accounts for changes in the victim's downloaded range
     * while we were requesting.
     * 
     * @param victim
     * @param slowestRange
     * @throws IOException
     */
    private void completeAssignGrey(DownloadWorker victim, Interval slowestRange) throws IOException {
        Interval newSlowestRange;
        int newStart;
        synchronized(victim.getDownloader()) {
            // if the victim died or was stopped while the thief was connecting, we can't steal
            if (!victim.getDownloader().isActive()) {
                LOG.debug("victim is no longer active");
                throw new NoSuchElementException();
            }
            
            // see how much did the victim download while we were exchanging headers.
            // it is possible that in that time some other worker died and freed his ranges, and
            // the victim has already been assigned some new ranges.  If that happened we don't steal.
            newSlowestRange = victim.getDownloadInterval();
            if (newSlowestRange.high != slowestRange.high) {
                if (LOG.isDebugEnabled())
                    LOG.debug("victim is now downloading something else "+
                            newSlowestRange+" vs. "+slowestRange);
                throw new NoSuchElementException();
            }
            
            if (newSlowestRange.low > slowestRange.low && LOG.isDebugEnabled()) {
                LOG.debug("victim managed to download "+(newSlowestRange.low - slowestRange.low)
                        +" bytes while stealer was connecting");
            }
            
            int myLow = _downloader.getInitialReadingPoint();
            int myHigh = _downloader.getAmountToRead() + myLow; // EXCLUSIVE
            
            // If the stealer isn't going to give us everything we need,
            // there's no point in stealing, so throw an exception and
            // don't steal.
            if( myHigh < slowestRange.high ) {
                if(LOG.isDebugEnabled()) {
                    LOG.debug("WORKER: not stealing because stealer " +
                            "gave a subrange.  Expected low: " + slowestRange.low +
                            ", high: " + slowestRange.high + ".  Was low: " + myLow +
                            ", high: " + myHigh);
                }
                throw new IOException();
            }
            
            newStart = Math.max(newSlowestRange.low,myLow);
            if(LOG.isDebugEnabled()) {
                LOG.debug("WORKER:"+
                        " picking stolen grey "
                        +newStart + "-"+slowestRange.high+" from ["+victim+"] to [" + this + "]");
            }
            
            
            // tell the victim to stop downloading at the point the thief 
            // can start downloading
            victim.getDownloader().stopAt(newStart);
        }
        
        // once we've told the victim where to stop, make our ranges release-able
        _downloader.startAt(newStart);
        _shouldRelease = true;
    }
    
    Interval getDownloadInterval() {
        HTTPDownloader downloader = _downloader;
        synchronized(downloader) {
            
            int start = Math.max(downloader.getInitialReadingPoint() + downloader.getAmountRead(),
                    downloader.getInitialWritingPoint());
            
            int stop = downloader.getInitialReadingPoint() + downloader.getAmountToRead();
            
            return new Interval(start,stop);
        }
    }
    
    /** Sets this worker as being part of or not part of a stealing operation. */
    private void setStealing(boolean stealing) {
        this._stealing = stealing;
    }
    
    /** Returns true if this worker is currently involved in a stealing operation. */
    boolean isStealing() {
        return _stealing;
    }
    
    /**
     * @return the httpdownloader that is going slowest.
     */
    private DownloadWorker findSlowestDownloader() {
        DownloadWorker slowest = null;
        final float ourSpeed = getOurSpeed();
        float slowestSpeed = ourSpeed;
        
        Set queuedWorkers = _manager.getQueuedWorkers().keySet();
        for (Iterator iter=_manager.getAllWorkers().iterator(); iter.hasNext();) {
            DownloadWorker worker = (DownloadWorker) iter.next();
            if(worker.isStealing())
                continue;
            
            if (queuedWorkers.contains(worker))
                continue;
            
            HTTPDownloader h = worker.getDownloader();
            
            if (h == null || h == _downloader || h.isVictim())
                continue;
            
            // if we don't have speed yet, steal from the first slow guy
            if (ourSpeed == UNKNOWN_SPEED) {
                if (worker.isSlow()) 
                    return worker;
            } else {
                // see if he is the slowest one
                float hisSpeed = 0;
                try {
                    h.getMeasuredBandwidth();
                    hisSpeed = h.getAverageBandwidth();
                } catch (InsufficientDataException ide) {
                    // we assume these guys would go almost as fast as we do, so we do not steal
                    // from them unless they are the last ones remaining
                    hisSpeed = Math.max(0f,ourSpeed - 0.1f);
                }
                
                if (hisSpeed < slowestSpeed) {
                    slowestSpeed = hisSpeed;
                    slowest = worker;
                }
            }
            
        }
        return slowest;
    }
    
    private float getOurSpeed() {
        if (_downloader == null)
            return UNKNOWN_SPEED;
        try {
            _downloader.getMeasuredBandwidth();
            return _downloader.getAverageBandwidth();
        } catch (InsufficientDataException bad) {
            return UNKNOWN_SPEED;
        }
    }
    
    /**
     * Returns true if the victim is going below minimum speed.
     * @return
     */
    boolean isSlow() {
        float ourSpeed = getOurSpeed();
        return ourSpeed < MIN_ACCEPTABLE_SPEED && ourSpeed != UNKNOWN_SPEED;
    }
    
    ////// various handlers for failure states of the assign process /////
    
    /**
     * no more ranges to download or no more people to steal from - finish download 
     */
    private ConnectionStatus handleNoMoreDownloaders() {
        _manager.addRFD(_rfd);
        return ConnectionStatus.getNoData();
    }
    
    /**
     * The file does not have such ranges 
     */
    private ConnectionStatus handleNoRanges() {
        //forget the ranges we are preteding uploader is busy.
        _rfd.setAvailableRanges(null);
        
        //if this RFD did not already give us a retry-after header
        //then set one for it.
        if(!_rfd.isBusy())
            _rfd.setRetryAfter(NO_RANGES_RETRY_AFTER);
        
        _rfd.resetFailedCount();                
        _manager.addRFD(_rfd);
        
        return ConnectionStatus.getNoFile();
    }
    
    private ConnectionStatus handleTryAgainLater() {
        //if this RFD did not already give us a retry-after header
        //then set one for it.
        if ( !_rfd.isBusy() ) {
            _rfd.setRetryAfter(RETRY_AFTER_NONE_ACTIVE);
        }
        
        //if we already have downloads going, then raise the
        //retry-after if it was less than the appropriate amount
        if(!_manager.getActiveWorkers().isEmpty() &&
                _rfd.getWaitTime(System.currentTimeMillis()) < RETRY_AFTER_SOME_ACTIVE)
            _rfd.setRetryAfter(RETRY_AFTER_SOME_ACTIVE);
        
        _manager.addRFD(_rfd);//try this rfd later
        
        _rfd.resetFailedCount();                
        return ConnectionStatus.getNoFile();
    }
    
    /**
     * The ranges exist in the file, but the remote host does not have them
     */
    private ConnectionStatus handleRangeNotAvailable() {
        _rfd.resetFailedCount();                
        _manager.informMesh(_rfd, true);
        //no need to add to files or busy we keep iterating
        return ConnectionStatus.getPartialData();
    }
    
    private ConnectionStatus handleFileNotFound() {
        _manager.informMesh(_rfd, false);
        return ConnectionStatus.getNoFile();
    }
    
    private ConnectionStatus handleNotSharing() {
        return handleFileNotFound();
    }
    
    private ConnectionStatus handleQueued(int position, int pollTime) {
        synchronized(_manager) {
            if(_manager.getActiveWorkers().isEmpty()) {
                if(_manager.isCancelled() || _manager.isPaused() ||  _interrupted)
                    return ConnectionStatus.getNoData(); // we were signalled to stop.
                _manager.setState(ManagedDownloader.REMOTE_QUEUED);
            }
            _rfd.resetFailedCount();                
            return ConnectionStatus.getQueued(position, pollTime);
        }
    }
    
    private ConnectionStatus handleProblemReadingHeader() {
        return handleFileNotFound();
    }
    
    private ConnectionStatus handleUnknownCode() {
        return handleFileNotFound();
    }
    
    private ConnectionStatus handleIO(){
        handleRFDFailure(_rfd);
        
        return ConnectionStatus.getNoFile();
    }
    
    //////// end handlers of various failure states ///////
    
    /**
     * interrupts this downloader.
     */
    void interrupt() {
        _interrupted = true;
        
        synchronized(_currentState) {
            if(_currentState.getCurrentState() == DownloadState.QUEUED)
                finishHttpLoop();
        }
        
        if(LOG.isDebugEnabled())
            LOG.debug("Stopping while state is: " + _currentState + ", this: " + toString());
        
        // If a downloader is set up, we don't need to deal
        // with the connector, since connecting has finished.
        if (_downloader != null) {
            _downloader.stop();
        } else {
            //Ensure that the ConnectObserver is cleaned up.
            DirectConnector observer = _connectObserver;
            if(observer != null) {
                Socket socket = observer.getSocket();
                // Make sure it immediately stops trying to connect.
                if(socket != null)
                    IOUtils.close(socket);
            }
        }
    }

    
    public RemoteFileDesc getRFD() {
        return _rfd;
    }
    
    HTTPDownloader getDownloader() {
        return _downloader;
    }
    
    public String toString() {
        return _workerName + "[" + _currentState + "] -> "+_rfd;  
    }
    
    /** Ensures this worker is finished and doesn't start again. */
    private void finishWorker() {
        _interrupted = true;
        _manager.workerFinished(this);
    }
    
    /**
     * Starts a new thread that will perform the download.
     * @param dl
     */
    private void startDownload(HTTPDownloader dl) {
        _downloader = dl;
        
        // If we should continue, then start the download.
        if(finishConnect()) {
            LOG.trace("Starting download");
            initializeAlternateLocations();
            httpLoop();
        } else {
            finishWorker();
        }
    }
    
    /**
     * Completes the http loop of this downloader, effectively
     * finishing its reign of downloading.
     */
    private void finishHttpLoop() {
        releaseRanges();
        _manager.removeQueuedWorker(this);
        _downloader.stop();
        finishWorker();
    }
    
    /**
     * A simple IOStateObserver that will increment state upon completion
     * and finish on close/shutdown, but offer the abillity for something
     * to be done prior to moving on in each case.
     */
    private abstract class State implements IOStateObserver {
        public final void handleIOException(IOException iox) {
            handleState(false);
            finishHttpLoop();
        }

        public final void handleStatesFinished() {
            handleState(true);
            incrementState(null);
        }

        public final void shutdown() {
            handleState(false);
            finishHttpLoop();
        }

        /** Handles per-state updating. */
        protected abstract void handleState(boolean success);
        
    }
    
    /**
     * A ConnectObserver for starting the download via a push connect.
     */
    private class PushConnector extends HTTPConnectObserver {
        private boolean forgetOnFailure;
        private boolean directConnectOnFailure;
        private PushDetails pushDetails;
        
        /**
         * Creates a new PushConnector.  If forgetOnFailure is true,
         * this will call _manager.forgetRFD(_rfd) if the push fails.
         * If directConnectOnFailure is true, this will attempt a direct
         * connection if the push fails.
         * Upon success, this will always start the download.
         * 
         * @param forgetOnFailure
         * @param directConnectOnFailure
         */
        PushConnector(boolean forgetOnFailure, boolean directConnectOnFailure) {
            this.forgetOnFailure = forgetOnFailure;
            this.directConnectOnFailure = directConnectOnFailure;
        }

        /**
         * Notification that the push succeeded.  Starts the download if the connection still exists.
         */
        public void handleConnect(Socket socket) {
            //LOG.debug(_rfd + " -- Handling connect from PushConnector");
            //HTTPDownloader dl = new HTTPDownloader(socket, _rfd, _commonOutFile, _manager instanceof InNetworkDownloader);
            HTTPDownloader dl = new HTTPDownloader(socket, _rfd, _commonOutFile, false);
            try {
               dl.connectTCP(0);
               DownloadStat.CONNECT_PUSH_SUCCESS.incrementStat();
            } catch(IOException iox) {
                //LOG.debug(_rfd + " -- IOX after starting connected from PushConnector.");
                DownloadStat.PUSH_FAILURE_LOST.incrementStat();
                failed();
                return;
            }
            
            startDownload(dl);
        }

        /** Notification that the push failed. */
        public void shutdown() {
            //LOG.debug(_rfd + " -- Handling shutdown from PushConnector");            
            DownloadStat.PUSH_FAILURE_NO_RESPONSE.incrementStat();
            failed();
        }
        
        /** Sets the details that will be used to unregister the push observer. */
        void setPushDetails(PushDetails details) {
            this.pushDetails = details;
        }
        
        /**
         * Possibly tells the manager to forget this RFD, cleans up various things,
         * and tells the manager to forget this worker.
         */
        private void failed() {            
            _manager.unregisterPushObserver(pushDetails, false);
            
            if(!directConnectOnFailure) {
                if(forgetOnFailure) {
                    _manager.forgetRFD(_rfd);
                }
                finishConnect();
                finishWorker();
            } else {
                connectDirectly(new DirectConnector(false));
            }
        }
    }
    
    /**
     * A ConnectObserver for starting the download via a direct connect.
     */
    private class DirectConnector extends HTTPConnectObserver {
        private long createTime = System.currentTimeMillis();
        private boolean pushConnectOnFailure;
        private Socket connectingSocket;
        private boolean shutdown;
        
        /**
         * Creates a new DirectConnection.  If pushConnectOnFailure is true,
         * this will attempt a push connection if the direct connect fails.
         * Upon success, this will always start a new download.
         *   
         * @param pushConnectOnFailure
         */
        DirectConnector(boolean pushConnectOnFailure) {
            this.pushConnectOnFailure = pushConnectOnFailure;
        }
        
        /**
         * Upon succesful connect, create the HTTPDownloader with the right socket, and proceed to continue
         * downloading.
         */
        public void handleConnect(Socket socket) {
            this.connectingSocket = null;
            
            //LOG.debug(_rfd + " -- Handling connect from DirectConnector");
            NumericalDownloadStat.TCP_CONNECT_TIME.addData((int) (System.currentTimeMillis() - createTime));
            DownloadStat.CONNECT_DIRECT_SUCCESS.incrementStat();
            //HTTPDownloader dl = new HTTPDownloader(socket, _rfd, _commonOutFile, _manager instanceof InNetworkDownloader);
            HTTPDownloader dl = new HTTPDownloader(socket, _rfd, _commonOutFile, false);
            try {
                dl.connectTCP(0); // already connected, timeout doesn't matter.
            } catch(IOException iox) {
                shutdown(); // if it immediately IOX's, try a push instead.
                return;
            }
            
            startDownload(dl);
        }

        /**
         * Upon unsuccesful connect, try using a push (if pushConnectOnFailure is true).
         */
        public void shutdown() {
            this.shutdown = true;
            this.connectingSocket = null;
            
            //LOG.debug(_rfd + " -- Handling shutdown from DirectConnnector");
            DownloadStat.CONNECT_DIRECT_FAILURES.incrementStat();
            if(pushConnectOnFailure) {
                connectWithPush(new PushConnector(false, false));
            } else {
                finishConnect();
                finishWorker();
            }
        }
        
        void setSocket(Socket socket) {
            this.connectingSocket = socket;
        }
        
        Socket getSocket() {
            return this.connectingSocket;
        }

        public boolean isShutdown() {
            return shutdown;
        }
    }
}
