/*
 * CallbackSink.java
 *
 * Created on October 7, 2003, 4:01 PM
 */

package condor.gahp.gt3;

import condor.gahp.*;

import org.globus.ogsa.client.managers.NotificationSinkManager;
import org.globus.ogsa.base.gram.ManagedJobPortType;
import java.util.List;
import java.util.Vector;


import org.globus.gsi.proxy.ProxyPolicyHandler;
import org.globus.ogsa.GridServiceException;
import org.globus.ogsa.ServiceData;
//import org.globus.ogsa.base.gram.Start;
import org.globus.ogsa.base.gram.ManagedJobPortType;
import org.globus.ogsa.base.gram.service.ManagedJobServiceLocator;
import org.globus.ogsa.base.gram.service.ManagedJobServiceGridLocator;
import org.globus.ogsa.base.gram.types.FaultType;
import org.globus.ogsa.base.gram.types.JobStateType;
import org.globus.ogsa.base.gram.types.JobStatusType;
import org.globus.ogsa.client.managers.NotificationSinkManager;
import org.globus.ogsa.handlers.GrimProxyPolicyHandler;
import org.globus.ogsa.impl.base.gram.utils.rsl.RslParser;
import org.globus.ogsa.impl.base.gram.utils.rsl.RslParseException;
import org.globus.ogsa.impl.base.gram.utils.rsl.RslParserFactory;
import org.globus.ogsa.impl.base.gram.utils.rsl.GramJobAttributes;
import org.globus.ogsa.impl.base.gram.utils.MemoryProfiler;
import org.globus.ogsa.impl.core.service.ServicePropertiesImpl;
import org.globus.ogsa.impl.security.authentication.Constants;
import org.globus.ogsa.impl.security.authorization.Authorization;
import org.globus.ogsa.impl.security.authorization.SelfAuthorization;
import org.globus.ogsa.NotificationSinkCallback;
import org.globus.ogsa.utils.AnyHelper;
import org.globus.ogsa.utils.GridServiceFactory;
import org.globus.ogsa.utils.QueryHelper;
import org.globus.ogsa.types.properties.PropertiesDetailType;
import org.globus.rsl.ParseException;
import org.gridforum.ogsi.ExtensibilityType;
import org.gridforum.ogsi.Factory;
//import org.gridforum.ogsi.FaultType;
import org.gridforum.ogsi.GridService;
import org.gridforum.ogsi.HandleType;
import org.gridforum.ogsi.LocatorType;
import org.gridforum.ogsi.OGSIServiceLocator;
import org.gridforum.ogsi.ServiceDataValuesType;
import org.gridforum.ogsi.TerminationTimeType;
import org.gridforum.ogsi.ExtendedDateTimeType;
import org.gridforum.ogsi.InfinityType;
import org.gridforum.ogsi.EntryType;
import org.gridforum.ogsi.EntryContentType;

/**
 *
 * @author  ckireyev
 *
 * This class is a wrapper around NoticicationSinkManager - 
 * the GT3 way of receiving callbacks
 *
 */
public class CallbackSink  implements CleanupStep {
    private String callbackId;
    private int requestId;
    private NotificationSinkManager notificationSinkManager;
    private List jobListeners = new Vector();
    private GahpInterface gahp;
    private boolean isInitialized = false;
    private Vector callbackIDs = new Vector();
    
    public static final String CALLBACK_SINK = "CALLBACK_SINK";
    
    /** Creates a new instance of CallbackSink */
    public CallbackSink(NotificationSinkManager manager, int requestId, GahpInterface gahp) {
        this.notificationSinkManager = manager;
        this.requestId = requestId;
        this.gahp = gahp;
    }
    
    public synchronized void addJobListener (
        ManagedJobServiceGridLocator serviceLocator,
        ManagedJobPortType service )
            throws org.globus.ogsa.GridServiceException, java.rmi.RemoteException
    {
        if (!isInitialized) {
            // I forget why this is needed but I think it is
            service.findServiceData(QueryHelper.getNamesQuery(
                            "ManagedJobUserIdLocal"));
            this.notificationSinkManager.setService(serviceLocator);
            isInitialized=true;
        }

        JobListener jobListener =     
            new JobListener(requestId,
                            serviceLocator.getGSR().getHandle().toString(),
                            gahp);
        

        jobListener.callbackId = 
            this.notificationSinkManager.addListener(
                "ManagedJobState",
                null,
                serviceLocator.getGSR().getHandle(),
                jobListener);
        callbackIDs.add (jobListener.callbackId);
    }
    
    public synchronized void doCleanup() {
        try {
            for (int i=0; i<callbackIDs.size(); i++) {
                this.notificationSinkManager.removeListener (
                    (String)callbackIDs.get(i));
            }
            callbackIDs.clear();
            
            this.notificationSinkManager.stopListening();
        }
        catch (Exception e) {
            e.printStackTrace(System.err);
        }
    }
    
    public static void storeCallbackSink (GahpInterface gahp, String id, CallbackSink sink) {
        gahp.storeObject (CALLBACK_SINK+id, sink);
    }
    
    public static CallbackSink getCallbackSink (GahpInterface gahp, String id) {
        return (CallbackSink)gahp.getObject (CALLBACK_SINK+id);
    }
}
