making sure all posted tasks are processed even after process terminated and dispose called by using sequential task executor instead of MergingUpdateQueue (IDEA-120167 Phantom eternal Ant task on make)

java/execution/impl/src/com/intellij/execution/junit/JUnitProcessHandler.java

@@ -20,7 +20,6 @@ import com.intellij.execution.configurations.GeneralCommandLine;
 import com.intellij.execution.junit2.segments.Extractor;
 import com.intellij.execution.process.OSProcessHandler;
 import com.intellij.execution.process.ProcessTerminatedListener;
-import com.intellij.openapi.util.Disposer;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.Reader;
@@ -71,17 +70,4 @@ public class JUnitProcessHandler extends OSProcessHandler {
     return processHandler;
   }
 
-  @Override
-  protected void notifyProcessTerminated(int exitCode) {
-    super.notifyProcessTerminated(exitCode);
-    Disposer.dispose(myOut);
-    Disposer.dispose(myErr);
-  }
-
-  @Override
-  protected void notifyProcessDetached() {
-    super.notifyProcessDetached();
-    Disposer.dispose(myOut);
-    Disposer.dispose(myErr);
-  }
 }

java/execution/impl/src/com/intellij/execution/junit2/segments/Extractor.java

@@ -18,45 +18,30 @@ package com.intellij.execution.junit2.segments;
 import com.intellij.execution.junit.SegmentedInputStreamReader;
 import com.intellij.execution.junit2.SegmentedInputStream;
 import com.intellij.execution.testframework.Printable;
-import com.intellij.openapi.Disposable;
 import com.intellij.openapi.diagnostic.Logger;
-import com.intellij.openapi.util.Disposer;
 import com.intellij.rt.execution.junit.segments.PacketProcessor;
-import com.intellij.util.ui.UIUtil;
-import com.intellij.util.ui.update.MergingUpdateQueue;
-import com.intellij.util.ui.update.Update;
+import com.intellij.util.concurrency.SequentialTaskExecutor;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.ide.PooledThreadExecutor;
 
+import javax.swing.*;
 import java.io.InputStream;
 import java.io.Reader;
 import java.nio.charset.Charset;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.Executor;
 
 /**
  * @author dyoma
  */
-public class Extractor implements Disposable {
+public class Extractor {
   private DeferredActionsQueue myFulfilledWorkGate = null;
   private final SegmentedInputStream myStream;
   private OutputPacketProcessor myEventsDispatcher;
   private static final Logger LOG = Logger.getInstance("#" + Extractor.class.getName());
-  private final MergingUpdateQueue myQueue = new MergingUpdateQueue("Test Extractor", 20, true, MergingUpdateQueue.ANY_COMPONENT);
-  private AtomicInteger myOrder = new AtomicInteger(-1);
+  private final Executor myQueue2 = new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE);
 
   public Extractor(@NotNull InputStream stream, @NotNull Charset charset) {
     myStream = new SegmentedInputStream(stream, charset);
-    myQueue.setPassThrough(false);//should be updated in awt thread
-  }
-
-  @Override
-  public void dispose() {
-    myQueue.flush();
-    UIUtil.invokeLaterIfNeeded(new Runnable() {
-      @Override
-      public void run() {
-        Disposer.dispose(myQueue);
-      }
-    });
   }
 
   public void setDispatchListener(final DispatchListener listener) {
@@ -67,7 +52,7 @@ public class Extractor implements Disposable {
     myFulfilledWorkGate = new DeferredActionsQueue() { //todo make it all later
       @Override
       public void addLast(final Runnable runnable) {
-        myQueue.queue(new MyUpdate(runnable, queue, myOrder.incrementAndGet()));
+        scheduleTask(queue, runnable);
       }
 
       @Override
@@ -100,6 +85,25 @@ public class Extractor implements Disposable {
     myStream.setEventsDispatcher(myEventsDispatcher);
   }
 
+  private void scheduleTask(final DeferredActionsQueue queue, final Runnable task) {
+    myQueue2.execute(new Runnable() {
+      public void run() {
+        // for some reason there is a requirement that this activity must be done from the swing thread
+        // will be blocking one of pooled threads here, which is ok
+        try {
+          SwingUtilities.invokeAndWait(new Runnable() {
+            public void run() {
+              queue.addLast(task);
+            }
+          });
+        }
+        catch (Throwable e) {
+          LOG.info("Task rejected: " + task, e);
+        }
+      }
+    });
+  }
+
   public OutputPacketProcessor getEventsDispatcher() {
     return myEventsDispatcher;
   }
@@ -108,30 +112,7 @@ public class Extractor implements Disposable {
     return new SegmentedInputStreamReader(myStream);
   }
 
-  public void addRequest(Runnable runnable, final DeferredActionsQueue queue) {
-    myQueue.queue(new MyUpdate(runnable, queue, myOrder.incrementAndGet()));
-  }
-
-  private static class MyUpdate extends Update {
-    private final Runnable myRunnable;
-    private final DeferredActionsQueue myQueue;
-    private final int myOrder;
-
-    public MyUpdate(Runnable runnable, DeferredActionsQueue queue, int order) {
-      super(runnable);
-      myRunnable = runnable;
-      myQueue = queue;
-      myOrder = order;
-    }
-
-    @Override
-    public void run() {
-      myQueue.addLast(myRunnable);
-    }
-
-    @Override
-    public int compareTo(Object o) {
-      return myOrder - ((MyUpdate)o).myOrder;
-    }
+  public void addRequest(final Runnable runnable, final DeferredActionsQueue queue) {
+    scheduleTask(queue, runnable);
   }
 }