//----------------------------------------------------------------------
//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 java.io.IOException;
import java.net.URI;
import java.util.Collection;

import org.apache.log4j.Logger;
import org.globus.cog.abstraction.coaster.service.LocalTCPService;
import org.globus.cog.abstraction.coaster.service.RegistrationManager;
import org.globus.cog.abstraction.interfaces.ExecutionService;
import org.globus.cog.abstraction.interfaces.JobSpecification;
import org.globus.cog.abstraction.interfaces.Service;
import org.globus.cog.abstraction.interfaces.Task;
import org.globus.cog.karajan.workflow.service.channels.ChannelContext;

public class JobQueue implements RegistrationManager {
    public static final Logger logger = Logger.getLogger(JobQueue.class);

    private QueueProcessor local, coaster;
    private Settings settings;
    private LocalTCPService localService;
    private ChannelContext clientChannelContext;

    public JobQueue(LocalTCPService localService) throws IOException {
        settings = new Settings();
        this.localService = localService;
    }

    public void start() {
        local = new LocalQueueProcessor();
        local.start();
    }

    public void enqueue(Task t) {
        Service s = t.getService(0);
        String jm = null;
        JobSpecification spec = (JobSpecification) t.getSpecification();
        if (s instanceof ExecutionService) {
            jm = ((ExecutionService) s).getJobManager();
        }
        if (spec.isBatchJob()) {
            if (logger.isInfoEnabled()) {
                logger.info("Job batch mode flag set. Routing through local queue.");
            }
        }
        QueueProcessor qp;
        if (s.getProvider().equalsIgnoreCase("coaster") && !spec.isBatchJob()) {
            if (logger.isInfoEnabled()) {
                logger.info("Adding task " + t + " to coaster queue");
            }
            qp = getQueueProcessor(settings.getWorkerManager());
        }
        else {
            if (logger.isInfoEnabled()) {
                logger.info("Adding task " + t + " to local queue");
            }
            qp = local;
        }
        qp.enqueue(t);
    }

    protected synchronized QueueProcessor getQueueProcessor(String name) {
        if (coaster == null) {
            coaster = newQueueProcessor(name);
            coaster.setClientChannelContext(clientChannelContext);
            coaster.start();
        }
        return coaster;
    }

    private QueueProcessor newQueueProcessor(String name) {
        if (name.equals("local")) {
            return new LocalQueueProcessor();
        }
        else if (name.equals("block")) {
            return newBlockQueueProcessor();
        }
        else if (name.equals("passive")) {
            return new PassiveQueueProcessor(settings, localService.getContact());
        }
        else {
            throw new IllegalArgumentException("No such queue processor: " + name);
        }
    }

    private AbstractQueueProcessor newBlockQueueProcessor() {
        BlockQueueProcessor bqp = new BlockQueueProcessor(settings);
        Collection<URI> addrs = settings.getLocalContacts(localService.getPort());
        if (addrs == null) {
            settings.setCallbackURI(localService.getContact());
        }
        else {
            settings.setCallbackURIs(addrs);
        }
        return bqp;
    }

    public Settings getSettings() {
        return settings;
    }

    public void shutdown() {
        local.shutdown();
        coaster.shutdown();
    }

    public void setClientChannelContext(ChannelContext channelContext) {
        if (this.clientChannelContext != null) {
            return;
        }
        this.clientChannelContext = channelContext;
        local.setClientChannelContext(channelContext);
        if (coaster != null) {
            coaster.setClientChannelContext(channelContext);
        }
    }

    public String nextId(String id) {
        return ((RegistrationManager) coaster).nextId(id);
    }

    public String registrationReceived(String id, String url, ChannelContext channelContext) {
        return ((RegistrationManager) coaster).registrationReceived(id, url, channelContext);
    }

    public QueueProcessor getCoasterQueueProcessor() {
        return coaster;
    }
}

@@ -1412,24 +1412,280 @@
 calService;%0A
+        Collection%3CURI%3E addrs = settings.getLocalContacts(localService.getPort());%0A        if (addrs == null) %7B%0A            settings.setCallbackURI(localService.getContact());%0A        %7D%0A        else %7B%0A            settings.setCallbackURIs(addrs);%0A        %7D%0A
     %7D%0A%0A    p
@@ -3260,24 +3260,25 @@
   return new
+ 
 BlockQueuePr
@@ -3277,32 +3277,40 @@
 kQueueProcessor(
+settings
 );%0A        %7D%0A   
@@ -3537,24 +3537,24 @@
  %22 + name);%0A
+
         %7D%0A  
@@ -3562,422 +3562,8 @@
  %7D%0A%0A
-    private AbstractQueueProcessor newBlockQueueProcessor() %7B%0A        BlockQueueProcessor bqp = new BlockQueueProcessor(settings);%0A        Collection%3CURI%3E addrs = settings.getLocalContacts(localService.getPort());%0A        if (addrs == null) %7B%0A            settings.setCallbackURI(localService.getContact());%0A        %7D%0A        else %7B%0A            settings.setCallbackURIs(addrs);%0A        %7D%0A        return bqp;%0A    %7D%0A%0A
     

make sure that the settings always have a non-null callback URI