//----------------------------------------------------------------------
//This code is developed as part of the Java CoG Kit project
//The terms of the license can be found at http://www.cogkit.org/license
//This message may not be removed or altered.
//----------------------------------------------------------------------

/*
 * Created on Feb 13, 2008
 */
package org.globus.cog.abstraction.coaster.service.job.manager;

import org.globus.cog.abstraction.interfaces.Task;
import org.globus.cog.karajan.util.Queue;

public abstract class AbstractQueueProcessor extends Thread implements QueueProcessor {
    private final Queue q;
    private Queue.Cursor cursor;
    private boolean shutdownFlag;
    private boolean wrap;

    public AbstractQueueProcessor(String name) {
        super(name);
        q = new Queue();
    }

    public void enqueue(Task t) {
        synchronized (q) {
            if (shutdownFlag) {
                throw new IllegalStateException("Queue is shut down");
            }
            q.enqueue(new AssociatedTask(t));
            q.notifyAll();
        }
    }

    public void shutdown() {
        synchronized(q) {
            shutdownFlag = true;
        }
    }

    protected boolean getShutdownFlag() {
        return shutdownFlag;
    }

    protected final Queue getQueue() {
        return q;
    }

    protected final AssociatedTask take() throws InterruptedException {
        return (AssociatedTask) q.take();
    }

    protected final boolean hasWrapped() {
        synchronized (q) {
            return wrap;
        }
    }

    protected final void remove() {
        synchronized (q) {
            if (cursor == null) {
                throw new IllegalThreadStateException(
                        "next() was never called");
            }
            else {
                cursor.remove();
            }
        }
    }

    protected int queuedTaskCount() {
        synchronized (q) {
            return q.size();
        }
    }
}

@@ -418,32 +418,125 @@
 anager;%0A%0Aimport 
+java.util.concurrent.BlockingQueue;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0A%0Aimport 
 org.globus.cog.a
@@ -565,50 +565,8 @@
 Task
-;%0Aimport org.globus.cog.karajan.util.Queue
 ;%0A%0Ap
@@ -674,48 +674,39 @@
 nal 
-Queue q;%0A    private Queue.Cursor cursor
+BlockingQueue%3CAssociatedTask%3E q
 ;%0A  
@@ -850,21 +850,51 @@
 q = new 
-Queue
+LinkedBlockingQueue%3CAssociatedTask%3E
 ();%0A    
@@ -934,39 +934,8 @@
 ) %7B%0A
-        synchronized (q) %7B%0A    
     
@@ -954,28 +954,24 @@
 downFlag) %7B%0A
-    
             
@@ -1029,34 +1029,26 @@
 %22);%0A        
-    %7D%0A    
+%7D%0A
         q.en
@@ -1045,23 +1045,21 @@
       q.
-enqueue
+offer
 (new Ass
@@ -1079,45 +1079,8 @@
 ));%0A
-            q.notifyAll();%0A        %7D%0A
     
@@ -1115,38 +1115,8 @@
 ) %7B%0A
-        synchronized(q) %7B%0A    
     
@@ -1140,26 +1140,16 @@
 = true;%0A
-        %7D%0A
     %7D%0A%0A 
@@ -1229,72 +1229,8 @@
  %7D%0A%0A
-    protected final Queue getQueue() %7B%0A        return q;%0A    %7D%0A%0A
     
@@ -1315,25 +1315,8 @@
 turn
- (AssociatedTask)
  q.t
@@ -1384,362 +1384,20 @@
     
-synchronized (q) %7B%0A            return wrap;%0A        %7D%0A    %7D%0A%0A    protected final void remove() %7B%0A        synchronized (q) %7B%0A            if (cursor == null) %7B%0A                throw new IllegalThreadStateException(%0A                        %22next() was never called%22);%0A            %7D%0A            else %7B%0A                cursor.remove();%0A            %7D%0A        %7D
+return wrap;
 %0A   
@@ -1442,39 +1442,8 @@
 ) %7B%0A
-        synchronized (q) %7B%0A    
     
@@ -1463,26 +1463,16 @@
 size();%0A
-        %7D%0A
     %7D%0A%7D%0A

use standard java blocking queue