package hudson.remoting;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * {@link CommandTransport} that implements the read operation in a synchronous fashion.
 * 
 * <p>
 * This class uses a thread to pump commands and pass them to {@link CommandReceiver}.
 *     
 * @author Kohsuke Kawaguchi
 */
public abstract class SynchronousCommandTransport extends CommandTransport {
    protected Channel channel;

    /**
     * Called by {@link Channel} to read the next command to arrive from the stream.
     */
    abstract Command read() throws IOException, ClassNotFoundException, InterruptedException;

    @Override
    public void setup(Channel channel, CommandReceiver receiver) {
        this.channel = channel;
        new ReaderThread(receiver).start();
    }

    private final class ReaderThread extends Thread {
        private int commandsReceived = 0;
        private int commandsExecuted = 0;
        private final CommandReceiver receiver;

        public ReaderThread(CommandReceiver receiver) {
            super("Channel reader thread: "+channel.getName());
            this.receiver = receiver;
        }

        @Override
        public void run() {
            final String name =channel.getName();
            try {
                while(!channel.isInClosed()) {
                    Command cmd = null;
                    try {
                        cmd = read();
                    } catch (EOFException e) {
                        IOException ioe = new IOException("Unexpected termination of the channel");
                        ioe.initCause(e);
                        throw ioe;
                    } catch (ClassNotFoundException e) {
                        LOGGER.log(Level.SEVERE, "Unable to read a command (channel " + name + ")",e);
                        continue;
                    } finally {
                        commandsReceived++;
                    }

                    receiver.handle(cmd);
                    commandsExecuted++;
                }
                closeRead();
            } catch (InterruptedException e) {
                LOGGER.log(Level.SEVERE, "I/O error in channel "+name,e);
                channel.terminate((InterruptedIOException) new InterruptedIOException().initCause(e));
            } catch (IOException e) {
                LOGGER.log(Level.SEVERE, "I/O error in channel "+name,e);
                channel.terminate(e);
            } catch (RuntimeException e) {
                LOGGER.log(Level.SEVERE, "Unexpected error in channel "+name,e);
                channel.terminate((IOException) new IOException("Unexpected reader termination").initCause(e));
                throw e;
            } catch (Error e) {
                LOGGER.log(Level.SEVERE, "Unexpected error in channel "+name,e);
                channel.terminate((IOException) new IOException("Unexpected reader termination").initCause(e));
                throw e;
            } finally {
                channel.pipeWriter.shutdown();
            }
        }
    }

    private static final Logger LOGGER = Logger.getLogger(SynchronousCommandTransport.class.getName());
}

@@ -107,32 +107,72 @@
 tedIOException;%0A
+import java.net.SocketTimeoutException;%0A
 import java.util
@@ -187,16 +187,16 @@
 .Level;%0A
-
 import j
@@ -562,16 +562,466 @@
 annel;%0A%0A
+    private static final String RDR_SOCKET_TIMEOUT_PROPERTY_NAME = %0A            SynchronousCommandTransport.class.getName() + %22.failOnSocketTimeoutInReader%22;%0A    %0A    /**%0A     * Enables the original aggressive behavior, when the channel reader gets %0A     * interrupted on any %7B@link SocketTimeoutException%7D.%0A     * @since TODO%0A     */%0A    private static boolean RDR_FAIL_ON_SOCKET_TIMEOUT = Boolean.getBoolean(RDR_SOCKET_TIMEOUT_PROPERTY_NAME);%0A    %0A
     /**%0A
@@ -1947,32 +1947,32 @@
           try %7B%0A
-
                 
@@ -1989,24 +1989,889 @@
 d = read();%0A
+                    %7D catch (SocketTimeoutException ex) %7B%0A                        if (RDR_FAIL_ON_SOCKET_TIMEOUT) %7B%0A                            LOGGER.log(Level.SEVERE, %22Socket timeout in the Synchronous channel reader.%22%0A                                    + %22 The channel will be interrupted, because %22 + RDR_SOCKET_TIMEOUT_PROPERTY_NAME %0A                                    + %22 is set%22, ex);%0A                            throw ex;%0A                        %7D%0A                        // Timeout happened during the read operation.%0A                        // It is not always fatal, because it may be caused by a long-running command%0A                        // If channel is not closed, it's OK to continue reading the channel%0A                        LOGGER.log(Level.WARNING, %22Socket timeout in the Synchronous channel reader%22, ex);%0A                        continue;%0A
             

- Make AsynchronousCommandTransport tolerant against Socket timeouts (#86)