// Copyright 2006 Google Inc.  All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.enterprise.connector.common;

import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * A WorkQueueThread executes work in the WorkQueue and calls the associated
 * callback method when the work is complete.
 */
public class WorkQueueThread extends Thread {
  private static final Logger LOGGER = 
    Logger.getLogger(WorkQueueThread.class.getName());

  private WorkQueue workQueue;
  private boolean isWorking;
  private boolean exit; // true if thread should exit

  public WorkQueueThread(String name, WorkQueue workQueue) {
    super(name);
    this.workQueue = workQueue;
    isWorking = false;
    exit = false;
  }

  public boolean isWorking() {
    return isWorking;
  }

  /**
   * Interrupt this Thread and have it exit.
   */
  public void interruptAndKill() {
    exit = true;
    interrupt();
  }

  public void run() {
    while (true) {
      WorkQueueItem item;
      synchronized (workQueue) {
        while (0 == workQueue.getWorkCount()) {
          try {
            workQueue.wait();
          } catch (InterruptedException ie) {
            if (exit) {
              // thread exits, for example, when shutdown of WorkQueue occurs
              LOGGER.log(Level.INFO,
                "Interrupted WorkQueueThread is exiting due to interrupt "
                  + "and kill.");
              return;
            } else {
              // if we aren't killing this thread, we go back to wait for more
              // work
              LOGGER.log(Level.INFO, "Interrupted WorkQueueThread is fine so" +
                    " will continue to wait for work.", ie);
              continue;
            }
          }
        }

        item = workQueue.removeWork();
      }

      item.setWorkQueueThread(this);
      try {
        workQueue.preWork(item);
        isWorking = true;
        item.doWork();
      } catch (RuntimeException e) {
        LOGGER.log(Level.WARNING, "WorkQueueThread work had problems: ", e);
        continue;
      } finally {
        isWorking = false;
        workQueue.postWork(item);
      }
    }
  }
}

@@ -935,17 +935,16 @@
 LOGGER =
- 
 %0A    Log
@@ -1489,24 +1489,38 @@
 le (true) %7B%0A
+      try %7B%0A  
       WorkQu
@@ -1533,16 +1533,18 @@
 m item;%0A
+  
       sy
@@ -1564,24 +1564,26 @@
 orkQueue) %7B%0A
+  
         whil
@@ -1624,30 +1624,34 @@
 %7B%0A          
+  
 try %7B%0A
+  
             
@@ -1664,24 +1664,26 @@
 eue.wait();%0A
+  
           %7D 
@@ -1728,16 +1728,18 @@
         
+  
 if (exit
@@ -1738,24 +1738,26 @@
 if (exit) %7B%0A
+  
             
@@ -1822,16 +1822,18 @@
  occurs%0A
+  
         
@@ -1877,16 +1877,18 @@
         
+  
 %22Interru
@@ -1990,16 +1990,18 @@
     
+  
 return;%0A
     
@@ -1996,16 +1996,18 @@
 return;%0A
+  
         
@@ -2033,16 +2033,18 @@
         
+  
 // if we
@@ -2118,16 +2118,18 @@
     
+  
 // work%0A
     
@@ -2124,16 +2124,18 @@
 // work%0A
+  
         
@@ -2224,18 +2224,16 @@
         
-  
 %22 will c
@@ -2271,32 +2271,34 @@
 ;%0A              
+  
 continue;%0A      
@@ -2287,24 +2287,26 @@
   continue;%0A
+  
             
@@ -2309,34 +2309,38 @@
     %7D%0A          
-%7D%0A
+  %7D%0A  
         %7D%0A%0A     
@@ -2330,24 +2330,26 @@
          %7D%0A%0A
+  
         item
@@ -2377,27 +2377,31 @@
 rk();%0A      
+  
 %7D%0A%0A
+  
       item.s
@@ -2425,16 +2425,18 @@
 (this);%0A
+  
       tr
@@ -2439,32 +2439,34 @@
   try %7B%0A        
+  
 workQueue.preWor
@@ -2474,32 +2474,34 @@
 (item);%0A        
+  
 isWorking = true
@@ -2498,24 +2498,26 @@
 ing = true;%0A
+  
         item
@@ -2529,24 +2529,26 @@
 rk();%0A      
+  
 %7D catch (Run
@@ -2562,24 +2562,26 @@
 eption e) %7B%0A
+  
         LOGG
@@ -2641,24 +2641,26 @@
 ems: %22, e);%0A
+  
         cont
@@ -2671,16 +2671,18 @@
 ;%0A      
+  
 %7D finall
@@ -2681,24 +2681,26 @@
 %7D finally %7B%0A
+  
         isWo
@@ -2710,24 +2710,26 @@
 ng = false;%0A
+  
         work
@@ -2746,24 +2746,137 @@
 Work(item);%0A
+        %7D%0A%0A      %7D catch (Throwable t) %7B%0A        LOGGER.log(Level.WARNING, %22WorkQueueThread has problems: %22, t);%0A
       %7D%0A    

Add try/catch in order to handle weird errors such as OutOfMemoryError. Without such a catch, our WorkQueueThreads may exit and no other threads will replace them. This problem can be triggered by attempting to push a sufficiently large file.