Changeset 18359 for gli/branches/rtl-gli

Timestamp:

2009-01-12T11:18:36+13:00 (15 years ago)

Author:

kjdon

Message:

updated the rtl-gli branch with files from trunk. Result of a merge 14807:18318

Location:

gli/branches/rtl-gli/src/org/greenstone/gatherer/download

Files:

• DownloadJob.java (modified) (9 diffs)
• DownloadScrollPane.java (modified) (6 diffs)

gli/branches/rtl-gli/src/org/greenstone/gatherer/download/DownloadJob.java

@@ -76 +76 @@
 
     private String download_url = "";
-
+    
     //    private String current_url;
     //    private String destination;
@@ -96 +96 @@
     public static int DEFINED_MAX  = 1;
     public static int UNDEFINED_MAX  = 2;
-
+    
+    // To prematurely terminate wget, we will need to use sockets and find a free port.
+    // We will look at a limited range of ports. This range will be reused (circular buffer)
+    private static final int PORT_BASE = 50000;
+    private static final int PORT_BLOCK_SIZE = 100;   
+    private static int nextFreePort = PORT_BASE;  // Keep track what port numbers we have checked for availability
+    int port;                                     // package access. The socket port number this instance of DownloadJob will use
+    
     private String mode = null;
    
     private String proxy_url;
-
+    
     /**
      */
@@ -169 +176 @@
     // The stop_start_button is used to alternately start or stop the
     // job. If the current state of the job is paused then this
-    // restart is logically equivelent to a resume.
+    // restart is logically equivalent to a resume.
     if(event.getSource() == progress.stop_start_button) {
         previous_state = state;
@@ -189 +196 @@
     }
 
+    /** Given a portnumber to check, returns true if it is available
+     * (if nothing's listening there already). */
+    public static boolean isPortAvailable(int portnum) {
+    Socket tmpSocket = null;
+    try {
+        tmpSocket = new Socket("localhost", portnum);
+        tmpSocket.close();
+        return false;
+
+    } catch(ConnectException ex){
+        // "Signals that an error occurred while attempting to connect a socket 
+        // to a remote address and port. Typically, the connection was refused
+        // remotely (e.g., no process is listening on the remote address/port)."
+        System.err.println("Port " + portnum + " not yet in use.");
+        tmpSocket = null;
+        return true;
+
+    } catch(Exception ex) { 
+        // includes BindException "Signals that an error occurred while attempting
+        // to bind a socket to a local address and port. Typically, the port is in
+        // use, or the requested local address could not be assigned."
+        tmpSocket = null;
+        return false;
+    }
+    }
+
+    /** Circular buffer. Modifies the value of nextFreePort (the buffer index). */
+    private void incrementNextFreePort() {
+    int offset = nextFreePort - PORT_BASE;
+    offset = (offset + 1) % PORT_BLOCK_SIZE;
+    nextFreePort = PORT_BASE + offset;
+    } 
 
     public void callDownload() {
-    
+
     ArrayList command_list = new ArrayList();
         if (Utility.isWindows()) {
@@ -202 +241 @@
     command_list.add("-cache_dir");
     command_list.add(Gatherer.getGLIUserCacheDirectoryPath());
+    // For the purposes of prematurely terminating wget from GLI (which creates a socket
+    // as a communication channel between GLI and Perl), it is important to tell the script
+    // that we're running as GLI. Because when running from the command prompt, it should
+    // not create this socket and do the related processing.
+    command_list.add("-gli");
     
     ArrayList all_arg = download.getArguments(true,false);
@@ -252 +296 @@
         }
         }
- 
         //System.out.println(newcmd);       
 
-        InputStreamReader isr = new InputStreamReader(prcs.getErrorStream());
-        BufferedReader br = new BufferedReader(isr);
-        // Capture the standard error stream and seach for two particular occurances.
+        // Can use the following if debugging WgetDownload.pm - Reads debug stmts from the perl process' STDIN stream
+        //(new PerlReaderThread(prcs)).start();
+
+        InputStream is = prcs.getInputStream();
+        BufferedReader reader = new BufferedReader(new InputStreamReader(is));
+
+        // To be able to stop Wget, we use sockets to communicate with the perl process that launched wget
+        if (mode.equals("Web") || mode.equals("MediaWiki")) { // wget download modes other than OAI
+        
+        // Need to find an available (unused) port within the range we're looking for to pass it
+        // the Perl child process, so that it may set up a listening ServerSocket at that port number
+        try {
+            boolean foundFreePort = false;
+            for(int i = 0; i < PORT_BLOCK_SIZE; i++) {
+
+            if(isPortAvailable(nextFreePort)) {
+                foundFreePort = true;
+                break;
+                
+            } else {
+                incrementNextFreePort();
+            }
+            }
+            
+            if(foundFreePort) {
+            // Free port number currently found becomes the port number of the socket that this
+            // DownloadJob instance will be connecting to when the user wants to prematurely stop Wget.
+            this.port = nextFreePort;
+            incrementNextFreePort();
+            
+            } else {
+            throw new Exception("Cannot find an available port in the range " 
+                        + PORT_BASE + "-" + (PORT_BASE+PORT_BLOCK_SIZE)
+                        + "\nwhich is necessary for forcibly terminating wget.");
+            }
+
+            // Communicate the chosen port for this DownloadJob instance to the perl process, so
+            // that it can set up a ServerSocket at that port to listen for any signal to terminate wget
+            OutputStream os = prcs.getOutputStream();
+            String p = ""+this.port+"\n";
+            System.err.println("Portnumber found: " + p);
+
+            os.write(p.getBytes());
+            os.close();
+          
+        } catch(Exception ex) {
+            System.err.println("Sent available portnumber " + this.port + " to process' outputstream.\nBut got exception: " + ex);
+        }
+        }
+        
+        BufferedReader br = new BufferedReader(new InputStreamReader(prcs.getErrorStream()));
+        // Capture the standard error stream and search for two particular occurrences.
         String line="";
         boolean ignore_for_robots = false;
         int max_download = DownloadJob.UNKNOWN_MAX;
-
-
+       
         while ((line = br.readLine()) != null && !line.trim().equals("<<Finished>>") && state != STOPPED) {
-
         if ( max_download == DownloadJob.UNKNOWN_MAX) {
             if(line.lastIndexOf("<<Defined Maximum>>") != -1) {
@@ -355 +445 @@
         }
         }
+
         if(state == STOPPED) {
-        isr.close();
-        prcs.destroy(); // This doesn't always work, but it's worth a try
+        boolean terminatePerlScript = true;
+        
+        // When GLI is working with wget-based download modes other than OAI (MediaWiki and Web 
+        // download) and the STOP button has been pressed, wget needs to be prematurely terminated.
+        // Only wget download modes Web and MediaWiki require the use of sockets to communicate
+        // with the perl script in order to get wget to terminate. Other download modes, including
+        // wgetdownload mode OAI, can terminate in the traditional manner: close process inputstream 
+        // and kill perl process. OAI launches many wgets. So that when the perl process is terminated,
+        // the currently running wget will finish off but other wgets are no longer launched. 
+        if(prcs != null && (mode.equals("Web") || mode.equals("MediaWiki"))) { 
+
+            // create a socket to the perl child process and communicate the STOP message
+            Socket clientSocket = null;
+            if(clientSocket == null) {
+            try {
+                clientSocket = new Socket("localhost", this.port); // connect to the port chosen for this DownloadJob instance
+                
+                BufferedReader clientReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+                String response = clientReader.readLine(); // see if we've been connected
+                System.err.println("Communicating with perl download script on port " + this.port 
+                           + "\nGot response from perl: " + response); 
+                
+                // Send the STOP signal
+                OutputStream os = clientSocket.getOutputStream();
+                String message = "<<STOP>>\n";
+                os.write(message.getBytes());
+                response = clientReader.readLine(); // see whether the stop signal has been received
+                System.err.println("GLI sent STOP signal to perl to terminate wget." 
+                           + "\nGot response from perl: " + response);
+
+                response = clientReader.readLine(); // see whether the perl script is ready to be terminated
+                System.err.println("Got another response from perl: " + response);
+                os.close();
+                
+                clientReader.close();
+                clientSocket.close(); // close the clientSocket (the Perl end will close the server socket that Perl opened)
+                clientReader = null;
+                clientSocket = null;
+                
+                if(response == null) {
+                terminatePerlScript = false;
+                }
+            } catch(IOException ex) {
+                System.err.println("Tried to communicate through client socket - port " + this.port + ", but got exception: " + ex);
+            } catch(Exception ex) {
+                System.err.println("Tried to open client socket, but got exception: " + ex);
+            }
+            }
+        }
+        
+        //prcs.getInputStream().close();
+        prcs.getErrorStream().close();
+        br.close();
+        br = null;
+        if(terminatePerlScript) {
+            prcs.destroy(); // This doesn't always work, but it's worth a try       
+            prcs = null;
+        }
+
+        // Notify the DownloadScrollPane which is waiting on this job to complete that we are ready 
+        synchronized(this) {
+            this.notify();
+        }
         }
-       
     }
     catch (Exception ioe) {
@@ -372 +523 @@
         previous_state = state;
         state = DownloadJob.COMPLETE;
-        
     }
     // refresh the workspace tree
     Gatherer.g_man.refreshWorkspaceTree(WorkspaceTree.DOWNLOADED_FILES_CHANGED);
-       
     }
 
@@ -464 +613 @@
     progress.updateProgress(current, expected);
     }
+
+    
+    // Inner thread class that reads from process downloadfrom.pl's errorstream
+    private class PerlReaderThread extends Thread {
+    Process prcs = null;
+
+    public PerlReaderThread(Process proc) {
+        this.prcs = proc;
+    }
+
+    public void run() {
+        try {
+        if(prcs != null) {
+            String message = null;
+            BufferedReader eReader = new BufferedReader(new InputStreamReader(prcs.getInputStream()));
+            while(prcs != null && (message = eReader.readLine()) != null) {
+            if(!message.equals("\n")) {
+                System.err.println("**** Perl STDOUT: " + message);
+            }
+            }
+            
+            if(prcs != null && eReader != null) {
+            eReader.close();
+            eReader = null;
+            System.err.println("**** Perl ENDed.");
+            }
+        }
+        } catch(Exception e) {
+        System.err.println("Thread - caught exception: " + e);
+        }
+    }
+    }
 }

gli/branches/rtl-gli/src/org/greenstone/gatherer/download/DownloadScrollPane.java

@@ -52 +52 @@
     extends Thread {
     
-    /** <i>true</i> if there is a task currently being carried out, <i>false</i> otherwise. */
-    private boolean busy = false;
     /** <i>true</i> if verbose debug messages should be displayed, <i>false</i> otherwise. */
     private boolean debug = false;
@@ -69 +67 @@
     private Vector job_queue;
     static final private boolean simple = true;
-    
+
+        
     public DownloadScrollPane() {
     job = null;
@@ -81 +80 @@
     public void deleteDownloadJob(DownloadJob delete_me) {
     if (delete_me == job) {
-        while(busy) {
-        }
-        job = null;
-    }
+        try {
+        // Wait for the job to finish cleaning up, before we can continue cleaning up here.
+        // But sometimes the job has completed (terminated naturally) and in that case there's
+        // nothing to wait for.
+        synchronized(delete_me) {
+            if (!delete_me.hasSignalledStop()) { // don't wait if DownloadJob.COMPLETED 
+            delete_me.wait();
+            }
+        }
+        } catch (Exception e) {
+        e.printStackTrace();
+        }       
+    }
+    // Close button pressed, get rid of this download's dedicated pane
+    finishedDownloadJob(delete_me, true);
+    }
+
+    /** To be called when a download job has terminated naturally or was prematurely stopped
+     *  via the close button. 
+     *  Gets rid of this download's pane with buttons and progress bar if prematurely stopped. */
+    protected void finishedDownloadJob(DownloadJob delete_me, boolean removeDisplay) {
     if (delete_me.hasSignalledStop()) {
-        list_pane.remove(delete_me.getProgressBar());
+        if(removeDisplay) {
+        list_pane.remove(delete_me.getProgressBar());
+        list_pane.remove(filler_pane);
+        }
         job_queue.remove(delete_me);
-        list_pane.remove(filler_pane);
         if(job_queue.size() > 0) {
         Dimension progress_bar_size = delete_me.getProgressBar().getPreferredSize();
@@ -131 +149 @@
 
     DebugStream.println("About to create a new job");
-
-
+    
         DownloadJob new_job = new DownloadJob(download, Configuration.proxy_pass, Configuration.proxy_user, this, mode, proxy_url);
     // Tell it to run as soon as possible
-
-      
 
     new_job.setState(DownloadJob.RUNNING);
@@ -165 +180 @@
     //list_pane.setAlignmentX(Component.LEFT_ALIGNMENT);
     list_pane.updateUI();
-    new_job = null;
+    new_job = null;         //job = (DownloadJob) job_queue.get(index);
+
     synchronized(this) {
         notify(); // Just incase its sleeping.
@@ -189 +205 @@
     while(true) {
         if(job_queue.size() > 0) {
-        int index = 0;
-        while(job_queue.size() > 0 && index < job_queue.size()) {
-            job = (DownloadJob) job_queue.get(index);
+        while(!job_queue.isEmpty()) { 
+            job = (DownloadJob) job_queue.firstElement(); 
+            
             if(job.getState() == DownloadJob.RUNNING) {
-            DebugStream.println("DownloadJob " + job.toString() + " Begun.");
-            busy = true;
+            DownloadJob delete_me = job;
+            String jobDisplayString = job.toString();
+            DebugStream.println("DownloadJob " + jobDisplayString + " Begun.");
+            System.err.println("DownloadJob " + job.port + " " + job.toString() + " Begun.");
             job.callDownload();
-            busy = false;
-            DebugStream.println("DownloadJob " + job.toString() + " complete.");
+            // Job is done. Ended naturally, don't get rid of this download's separate display panel
+            finishedDownloadJob(delete_me, false); 
+            System.err.println("DownloadJob " + jobDisplayString + " complete.");
+            DebugStream.println("DownloadJob " + jobDisplayString + " complete."); // by this point job is null!
             job = null;
-            }
-            index++;
+            delete_me = null;
+            }
         }
         try {