Changeset 1304

Timestamp:

Mar 3, 2011, 9:19:00 AM (11 years ago)

Author:

Nicklas Nordborg

Message:

Added support for aborting a torrent upload. Added support for resuming uploads in case of a server restart.

Location:

extensions/net.sf.basedb.torrent/trunk

Files:

• META-INF/extensions.xml (modified) (1 diff)
• resources/bittorrent.png (added)
• resources/upload_torrent.jsp (modified) (1 diff)
• src/main/net/sf/basedb/clients/torrent/service/TorrentManager.java (modified) (14 diffs)
• src/main/net/sf/basedb/clients/torrent/service/TorrentService.java (modified) (13 diffs)

extensions/net.sf.basedb.torrent/trunk/META-INF/extensions.xml

@@ -60 +60 @@
       <factory-class>net.sf.basedb.clients.web.extensions.toolbar.FixedButtonFactory</factory-class>
       <parameters>
-        <title>Upload torrent</title>
-        <onClick>Main.openPopup('$HOME$/upload_torrent.jsp?ID=' + getSessionId(), 'UploadTorrent', 600, 400)</onClick>
+        <title>Upload torrent&amp;hellip;</title>
+        <onClick>Main.openPopup('$HOME$/upload_torrent.jsp?ID=' + getSessionId(), 'UploadTorrent', 640, 480)</onClick>
+        <icon>~/bittorrent.png</icon>
       </parameters>
     </action-factory>

extensions/net.sf.basedb.torrent/trunk/resources/upload_torrent.jsp

@@ -70 +70 @@
       <table class="form" cellspacing=0>
       <tr>
-        <td class="prompt">Directory</td>
+        <td class="prompt">Save in</td>
         <td><%=HTML.encodeTags(directory.getPath().toString())%></td>
       </tr>

extensions/net.sf.basedb.torrent/trunk/src/main/net/sf/basedb/clients/torrent/service/TorrentManager.java

@@ -31 +31 @@
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.io.Writer;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.Properties;
 
 import org.slf4j.Logger;
@@ -40 +45 @@
 import net.sf.basedb.core.Directory;
 import net.sf.basedb.core.Job;
+import net.sf.basedb.core.signal.Signal;
+import net.sf.basedb.core.signal.SignalException;
+import net.sf.basedb.core.signal.SignalHandler;
 import net.sf.basedb.util.ChainedProgressReporter;
 import net.sf.basedb.util.FileUtil;
@@ -56 +64 @@
 */
 public class TorrentManager 
+  implements SignalHandler
 {
   
   private static final Logger log = LoggerFactory.getLogger(TorrentManager.class);
   
+  private final Collection<Signal> supportedSignals;
   private final TorrentService service;
   private final File workDir;
   private final String name;
   
-  private TorrentState state;
+  private volatile TorrentState state;
   private int jobId;
   private Torrent torrent;
+  private volatile Thread workerThread;
+  private volatile Signal receivedSignal;
   
   /**
@@ -78 +90 @@
     this.name = name;
     this.state = TorrentState.INITIAL;
-  }
-
+    this.supportedSignals = Collections.unmodifiableList(Arrays.asList(Signal.ABORT, Signal.SHUTDOWN));
+  }
+
+  /**
+    Create a new torrent manager for an already existing job. The manager will be
+    set in READY_TO_DOWNLOAD state and the download should automatically be
+    resumed where it was aborted. If the download is already completed this will be
+    detected and the files will be uploaded to BASE.
+  */
+  TorrentManager(TorrentService service, File workDir, String name, int jobId)
+  {
+    this(service, workDir, name);
+    this.jobId = jobId;
+    this.state = TorrentState.READY_TO_DOWNLOAD;
+  }
+  
+  /*
+    From the SignalHandler interface
+    --------------------------------
+  */
+  @Override
+  public Collection<Signal> getSupportedSignals() 
+  {
+    return supportedSignals;
+  }
+
+  @Override
+  public void handleSignal(Signal signal) 
+  {
+    Thread worker = workerThread;
+    if (signal == Signal.ABORT)
+    {
+      if (worker != null && worker.isAlive()) 
+      {
+        // Interrupt the worker thread and assume that it takes care
+        // of exception handling
+        receivedSignal = signal;
+        worker.interrupt();
+      }
+      else
+      {
+        // Close, report an error to BASE and cleanup.
+        close(new SignalException("Aborted by user."));
+      }
+    }
+    else if (signal == Signal.SHUTDOWN)
+    {
+      // We need to notify the worker thread
+      if (worker != null && worker.isAlive())
+      {
+        receivedSignal = signal;
+        worker.interrupt();
+      }
+    }
+  }
+
+  @Override
+  public boolean supports(Signal signal) 
+  {
+    return supportedSignals.contains(signal);
+  }
+  // ------------------------------------
   /**
     Get the current state of the torrent.
@@ -217 +289 @@
   */
   public void queueDownload()
+    throws IOException
   {
     if (getState() != TorrentState.INITIAL)
@@ -235 +308 @@
     }
   
+    writeProperties();
+    
     log.debug("Torrent manager '" + getName() + "' is READY_TO_DOWNLOAD");
     this.state = TorrentState.READY_TO_DOWNLOAD;
   }
 
+  private void writeProperties()
+    throws IOException
+  {
+    // Create job.properties file to store info about the related job
+    Writer writer = new PrintWriter(new File(getWorkDir(), "job.properties"), "UTF-8");
+    try
+    {
+      Properties p = new Properties();
+      p.setProperty("job.id", Integer.toString(jobId));
+      p.setProperty("original-filename", getName());
+      p.store(writer, null);
+    }
+    finally
+    {
+      if (writer != null) writer.close();
+    }
+  }
+
+  
   /**
     Called by the TorrentService when the download has been started.
@@ -269 +363 @@
     downloading, nothing is done.
   */
-  public void reportDownloadProgress(DbControl dc)
+  public synchronized void reportDownloadProgress(DbControl dc)
   {
     TorrentState state = getState();
@@ -297 +391 @@
   
   /**
-    Copy the downloaded files to BASE. 
-  */
-  public void copyToBase(DbControl dc)
+    Copy the downloaded files to BASE. This method is synchronized but may
+    be interrupted by calling {@link Thread#interrupt()} on the worker
+    thread. If the torrent hasn't been completely download nothing
+    is done.
+  */
+  public synchronized void copyToBase(DbControl dc)
     throws IOException
   {
@@ -305 +402 @@
     if (state != TorrentState.DOWNLOAD_COMPLETE) return;
     
-    this.state = TorrentState.UPLOADING;
-    File downloadDir = getDownloadDir();
-    Job job = getJob(dc);
-    Directory uploadDir = (Directory)job.getParameterValue("directory");
-    ChainedProgressReporter progress = new ChainedProgressReporter(job.getProgressReporter(null));
-    progress.setRange(90, 100);
-    List files = FileUtil.uploadFiles(dc, uploadDir, downloadDir, null, true, progress);
-    dc.refreshItem(job); // Needed since the progress reporter has modified the state in the database
-    job.doneOk(files.size() + " file(s) uploaded successfully");
-  }
-  
-  public void copyToBaseAsync()
+    try
+    {
+      this.state = TorrentState.UPLOADING;
+      this.workerThread = Thread.currentThread();
+
+      File downloadDir = getDownloadDir();
+      Job job = getJob(dc);
+      Directory uploadDir = (Directory)job.getParameterValue("directory");
+      ChainedProgressReporter progress = new ChainedProgressReporter(job.getProgressReporter(null));
+      progress.setRange(90, 100);
+      List files = FileUtil.uploadFiles(dc, uploadDir, downloadDir, null, true, progress);
+      dc.refreshItem(job); // Needed since the progress reporter has modified the state in the database
+      job.doneOk(files.size() + " file(s) uploaded successfully");
+    }
+    finally
+    {
+      this.workerThread = null;
+    }
+  }
+  
+  /**
+    Copy the downloaded files to BASE asynchronoulsy. This method will start a
+    new thread that calls {@link #copyToBase(DbControl)} and then immediately
+    return. To abort the upload set the interrupt flag on the thread using
+    {@link Thread#interrupt()}, or send the {@link Signal#ABORT} to
+    {@link #handleSignal(Signal)}.
+  */
+  public Thread copyToBaseAsync()
   {
     Thread t = new Thread(new Runnable()
@@ -332 +445 @@
         catch (Exception ex)
         {
-          close(ex);
+          // Is it an error or ABORT signal?
+          if (receivedSignal == null || receivedSignal == Signal.ABORT)
+          {
+            close(ex);
+          }
+          else if (receivedSignal == Signal.SHUTDOWN)
+          {
+            // Do nothing
+          }
         }
         finally
@@ -341 +462 @@
       }
     });
-    
     t.start();
+    return t;
   }
   
   /**
     Close this torrent manager and remove all temporary working files.
-    This method can be called any time, but it is intentded to be used
+    This method can be called any time, but it is intended to be used
     after a successful download and upload of the torrent files. If
     the download is being aborted due to an error, use
     {@link #close(Throwable)} instead.
   */
-  public void close()
+  public synchronized void close()
   {
     log.debug("Closing torrent manager '" + getName() + "'");
     
     // Remove the torrent from the service
-    service.remove(this);
+    service.removeTorrentManager(this);
     
     // Remote working directory
@@ -370 +491 @@
     working files will be deleted.
   */
-  public void close(Throwable t)
+  public synchronized void close(Throwable t)
   {
     log.debug("Closing torrent manager due to an error '" + getName() + "'", t);
@@ -378 +499 @@
   
     // Remove the torrent from the service
-    service.remove(this);
+    service.removeTorrentManager(this);
     
     // Remote working directory
@@ -420 +541 @@
   }
 
-  
 }

extensions/net.sf.basedb.torrent/trunk/src/main/net/sf/basedb/clients/torrent/service/TorrentService.java

@@ -29 +29 @@
 
 import java.io.File;
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -46 +48 @@
 import net.sf.basedb.core.SessionControl;
 import net.sf.basedb.core.Application.Pinger;
+import net.sf.basedb.core.signal.LocalSignalReceiver;
 import net.sf.basedb.util.FileUtil;
 import net.sf.basedb.util.Values;
@@ -111 +114 @@
   private SessionControl sc;
   private Pinger pinger;
+  private LocalSignalReceiver signalReciever;
   private volatile TimerTask serviceTask;
   
@@ -189 +193 @@
       this.btClient = tmpClient;
     this.torrentManagers = Collections.synchronizedList(new ArrayList<TorrentManager>());
+    this.signalReciever = new LocalSignalReceiver();
+    this.signalReciever.init("torrent-service:0");
+    
+    loadExistingTorrents();
+          
     this.isRunning = true;
     log.info("Bittorrent Download Service has been started");
@@ -204 +213 @@
     isRunning = false;
 
+    if (signalReciever != null) signalReciever.close(1000);
+    signalReciever = null;
     if (serviceTask != null) serviceTask.cancel();
     serviceTask = null;
@@ -218 +229 @@
   }
 
+  private void loadExistingTorrents()
+  {
+    // Iterate all subdirs in workDir
+    for (File subDir : workDir.listFiles())
+    {
+      if (!subDir.isDirectory()) continue; // with next file
+      
+      File jobProperties = new File(subDir, "job.properties");
+      if (!jobProperties.exists()) continue; // with next file
+      
+      Properties p = new Properties();
+      try
+      {
+        p.load(new InputStreamReader(new FileInputStream(jobProperties), "UTF-8"));
+      }
+      catch (Exception ex)
+      {
+        log.warn("Could not load " + jobProperties, ex);
+      }
+      int jobId = Values.getInt(p.getProperty("job.id"));
+      String name = p.getProperty("original-filename");
+      if (jobId > 0 && name != null)
+      {
+        log.debug("Resuming download: " + name + "; jobId=" +jobId);
+        registerTorrentManager(new TorrentManager(this, subDir, name, jobId));
+      }
+    }
+  }
+  
   /**
     Safely close the session control. Catches all exceptions but log
@@ -268 +308 @@
     if (!isRunning()) throw new IllegalStateException("The service has not been started.");
     
+    log.debug("Creating torrent manager for torrent: " + torrentName);
     // Create a sub-dir for this torrent
     String tmpDir = torrentName;
@@ -278 +319 @@
     // Create a Torrent manager
     TorrentManager manager = new TorrentManager(this, subDir, torrentName);
+    registerTorrentManager(manager);
+    log.debug("Created torrent manager for torrent: " + torrentName);
+    return manager;
+  }
+
+  /**
+    Register a new torrent manager
+  */
+  private void registerTorrentManager(TorrentManager manager)
+  {
     torrentManagers.add(manager);
-    
     // Create the service task if needed
       if (serviceTask == null)
@@ -285 +335 @@
           serviceTask = Application.getScheduler().schedule(new TorrentServiceStateCheckerTask(this), 10000, 10000, false);
       }
-    return manager;
-  }
-
-  /**
-    Remove an existing torrent manager. If this is the last one,
-    the service task will be closed.
-  */
-  synchronized void remove(TorrentManager manager)
-  {
+  }
+  
+  /**
+    Remove an existing torrent manager. 
+  */
+  synchronized void removeTorrentManager(TorrentManager manager)
+  {
+    log.debug("Stopping torrent manager: " + manager.getName() + "; jobId=" + manager.getJobId());
     try
     {
       btClient.stopTorrent(manager.getTorrent());
+      log.debug("Stopped torrent manager: " + manager.getName() + "; jobId=" + manager.getJobId());
     }
     catch (Exception ex)
@@ -303 +353 @@
     }
     torrentManagers.remove(manager);
+    
+    // Cancel the service task if there are no more torrents
     if (torrentManagers.size() == 0)
     {
@@ -334 +386 @@
   void startDownload(TorrentManager manager)
   {
-
+    log.debug("Starting download: " + manager.getName() + "; jobId=" + manager.getJobId());
     // Start the job + register signal receiver
     DbControl dc = sc.newDbControl();
@@ -341 +393 @@
       // Load the job and register it as started
       Job job = manager.getJob(dc);
-      //String handlerId = signalReciever.registerSignalHandler(manager);
-      //job.setSignalTransporter(signalReciever.getSignalTransporterClass(), handlerId);
-      job.start("Starting download...", "torrent-service@" + Application.getHostName(), null);
+      String handlerId = signalReciever.registerSignalHandler(manager);
+      job.setSignalTransporter(signalReciever.getSignalTransporterClass(), handlerId);
+      if (job.getStatus() != Job.Status.EXECUTING)
+      {
+        job.start("Starting download...", "torrent-service@" + Application.getHostName(), null);
+      }
       dc.commit();
     }
@@ -361 +416 @@
       // Set torrent manager state to DOWNLOADING
       manager.setTorrent(torrent);
+      log.debug("Started download: " + manager.getName() + "; jobId=" + manager.getJobId());
     }
     catch (Exception ex)