//----------------------------------------------------------------------
//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 15, 2008
 */
package org.globus.cog.abstraction.coaster.service;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Map;

import org.apache.log4j.Logger;
import org.globus.cog.karajan.workflow.service.ConnectionHandler;
import org.globus.cog.karajan.workflow.service.RequestManager;
import org.globus.cog.karajan.workflow.service.Service;
import org.globus.cog.karajan.workflow.service.ServiceContext;
import org.globus.cog.karajan.workflow.service.channels.ChannelContext;
import org.globus.cog.karajan.workflow.service.channels.ChannelException;
import org.globus.cog.karajan.workflow.service.channels.ChannelManager;
import org.globus.cog.karajan.workflow.service.channels.KarajanChannel;
import org.globus.cog.karajan.workflow.service.channels.TCPChannel;
import org.globus.common.CoGProperties;
import org.globus.net.PortRange;

public class LocalTCPService implements Registering, Service, Runnable {
    public static final Logger logger = Logger.getLogger(LocalTCPService.class);

    public static final int TCP_BUFSZ = 32768;

    private RegistrationManager registrationManager;
    private final TCPBufferManager buffMan;

    private ServerSocketChannel channel;
    private int port;
    
    private RequestManager requestManager;
    private ServiceContext context = new ServiceContext(this);
    
    private Thread serverThread;
    
    private URI contact;

    public LocalTCPService(RequestManager rm) throws IOException {
        this(rm, 0);
    }

    public LocalTCPService(RequestManager rm, int port) throws IOException {
        setRequestManager(rm);
        buffMan = new TCPBufferManager();
        this.port = port;
    }

    public String registrationReceived(String blockid, String url, 
           KarajanChannel channel, Map<String, String> options) throws ChannelException {
        if (logger.isInfoEnabled()) {
            logger.info("Received registration: blockid = " +
                        blockid + ", url = " + url);
        }
        ChannelContext cc = channel.getChannelContext();
        cc.getChannelID().setLocalID(blockid);
        String wid = registrationManager.nextId(blockid);
        cc.getChannelID().setRemoteID(wid);
        ChannelManager.getManager().registerChannel(cc.getChannelID(), channel);
        registrationManager.registrationReceived(blockid, wid, url, cc, options);
        return wid;
    }

    public void unregister(String id) {
        throw new UnsupportedOperationException();
    }

    public RegistrationManager getRegistrationManager() {
        return registrationManager;
    }

    public void setRegistrationManager(RegistrationManager workerManager) {
        this.registrationManager = workerManager;
    }
    
    protected void setRequestManager(RequestManager requestManager) {
        this.requestManager = requestManager;
    }
    
    public ServiceContext getContext() {
        return context;
    }
   
    public boolean bindPort() {
        try {
            channel.socket().bind(new InetSocketAddress(port));
            return true;
        }
        catch (Exception e) {
            return false;
        }
    }
   
    public void start() throws IOException {
        channel = ServerSocketChannel.open();
        channel.configureBlocking(true);
        
        /* When GLOBUS_TCP_PORT_RANGE is defined, find an acceptable port in that range */
        String globusTCPPortRange = System.getenv("GLOBUS_TCP_PORT_RANGE");
        String rangeValues[] = {"0", "0"};

        if(globusTCPPortRange != null) {
            rangeValues = globusTCPPortRange.split(",");
        }
   
        port = Integer.valueOf(rangeValues[0]);
        while(!bindPort()) {
            port++;
            if(port > Integer.valueOf(rangeValues[1])) {
                String msg = "Unable to find an available port";
                if(globusTCPPortRange != null)
                    msg += " in the range of " + rangeValues[0] + " to " + rangeValues[1];
                throw new IOException(msg);
            }
        }

        try {
            if (serverThread == null) {
                serverThread = new Thread(this);
                serverThread.setDaemon(true);
                serverThread.setName("Local TCP service");
                serverThread.start();
            }
        }
        catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
    public void run() {
        while(true) {
            try {
                SocketChannel c = channel.accept();
                handleConnection(c.socket());
            }
            catch (IOException e) {
                logger.warn("Accept() failed", e);
                return;
            }
        }
    }

    protected void handleConnection(Socket socket) {
        try {
            buffMan.addSocket(socket);
            socket.setSoLinger(false, 0);
            socket.setTcpNoDelay(true);
        }
        catch (SocketException e) {
            e.printStackTrace();
        }
        try {
            new WorkerConnectionHandler(this, buffMan.wrap(socket), requestManager).start();
        }
        catch (Exception e) {
            logger.warn("Could not start worker connection", e);
        }
    }
    
    public synchronized URI getContact() {
        if (contact == null) {
            String host = CoGProperties.getDefault().getIPAddress();
            if (host == null) {
                try {
                    host = InetAddress.getLocalHost().getHostAddress();
                }
                catch (UnknownHostException e) {
                    host = "127.0.0.1";
                }
            }
            try {
                contact = new URI("http://" + host + ":" + getPort());
            }
            catch (URISyntaxException e) {
                logger.warn("Cannot build local service contact URI", e);
            }
        }
        return contact;
    }
    
    public int getPort() {
        if (port == 0) {
            return channel.socket().getLocalPort();
        }
        else {
            return port;
        }
    }

    public boolean isRestricted() {
        return false;
    }

    public void irrecoverableChannelError(KarajanChannel channel, Exception e) {
        System.err.println("Irrecoverable channel exception: " + e.getMessage());
        System.exit(2);
    }
    
    private static int idSeq = 1;
    
    private synchronized static int nextId() {
        return idSeq++;
    }

    private static class WorkerConnectionHandler extends ConnectionHandler {
        public WorkerConnectionHandler(Service service, Socket socket, RequestManager requestManager)
                throws IOException {
            super(socket, new WorkerChannel(socket, requestManager, 
                new ChannelContext("worker-" + nextId(), service)), requestManager);
        }
    }
    
    private static class WorkerChannel extends TCPChannel {
        public WorkerChannel(Socket socket, RequestManager requestManager,
                ChannelContext channelContext) throws IOException {
            super(socket, requestManager, channelContext);
        }

        @Override
        protected boolean clientControlsHeartbeats() {
            // Reverse the heartbeat direction for coaster workers because it makes detection of lost
            // connections quicker (at least when heartbeats are considered lost if not seen for
            // x > interval).
            return false;
        }
    }    
}

@@ -4114,16 +4114,50 @@
  = %7B
-%220%22, %220%22
+String.valueOf(port), String.valueOf(port)
 %7D;%0A%0A
@@ -4189,24 +4189,37 @@
 ange != null
+ && port == 0
 ) %7B%0A        

Fix for when coaster-service -localport is used