// Copyright (c) 2009 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

package org.chromium.sdk.internal.transport;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.chromium.sdk.ConnectionLogger;
import org.chromium.sdk.internal.transport.Message.MalformedMessageException;

/**
 * The low-level network agent handling the reading and writing of Messages
 * using the debugger socket.
 *
 * This class is thread-safe.
 */
public class SocketConnection implements Connection {

  /**
   * A thread that can be gracefully interrupted by a third party.
   * <p>
   * Unfortunately there is no standard way of interrupting I/O in Java. See Bug #4514257
   * on Java Bug Database (http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4514257).
   */
  private static abstract class InterruptibleThread extends Thread {

    protected volatile boolean isTerminated = false;

    InterruptibleThread(String name) {
      super(name);
    }

    @Override
    public synchronized void start() {
      this.isTerminated = false;
      super.start();
    }

    @Override
    public synchronized void interrupt() {
      this.isTerminated = true;
      super.interrupt();
    }
  }

  /**
   * Character encoding used in the socket data interchange.
   */
  private static final String SOCKET_CHARSET = "UTF-8";

  /**
   * A thread writing client-supplied messages into the debugger socket.
   */
  private class WriterThread extends InterruptibleThread {

    private final BufferedWriter writer;

    public WriterThread(BufferedWriter writer) {
      super("WriterThread");
      this.writer = writer;
    }

    @Override
    public void run() {
      while (!isTerminated && isAttached.get()) {
        try {
          handleOutboundMessage(outboundQueue.take());
        } catch (InterruptedException e) {
          // interrupt called on this thread, exit on isTerminated
        }
      }
    }

    private void handleOutboundMessage(Message message) {
      try {
        LOGGER.log(Level.FINER, "-->{0}", message);
        message.sendThrough(writer);
      } catch (IOException e) {
        SocketConnection.this.shutdown(e, false);
      }
    }
  }

  private static abstract class MessageItem {
    abstract void report(NetListener listener);
    abstract boolean isEos();
  }
  private static final MessageItem EOS = new MessageItem() {
    @Override
    void report(NetListener listener) {
      LOGGER.log(Level.FINER, "<--EOS");
      listener.eosReceived();
    }
    @Override
    boolean isEos() {
      return true;
    }
  };
  private static class RegularMessageItem extends MessageItem {
    private final Message message;
    RegularMessageItem(Message message) {
      this.message = message;
    }
    @Override
    void report(NetListener listener) {
      LOGGER.log(Level.FINER, "<--{0}", message);
      listener.messageReceived(message);
    }
    @Override
    boolean isEos() {
      return false;
    }
  }

  /**
   * A thread reading data from the debugger socket.
   */
  private class ReaderThread extends InterruptibleThread {

    private final BufferedReader reader;
    private final Writer handshakeWriter;

    public ReaderThread(BufferedReader reader, Writer handshakeWriter) {
      super("ReaderThread");
      this.reader = reader;
      this.handshakeWriter = handshakeWriter;
    }

    @Override
    public void run() {
      Exception breakException;
      try {
        /** The thread that dispatches the inbound messages (to avoid queue growth.) */
        startResponseDispatcherThread();

        if (connectionLogger != null) {
          connectionLogger.start();
        }

        handshaker.perform(reader, handshakeWriter);

        startWriterThread();

        while (!isTerminated && isAttached.get()) {
          Message message;
          try {
            message = Message.fromBufferedReader(reader);
          } catch (MalformedMessageException e) {
            LOGGER.log(Level.SEVERE, "Malformed protocol message", e);
            continue;
          }
          if (message == null) {
            LOGGER.fine("End of stream");
            break;
          }
          inboundQueue.add(new RegularMessageItem(message));
        }
        breakException = null;
      } catch (IOException e) {
        breakException = e;
      } finally {
        inboundQueue.add(EOS);
      }
      if (!isInterrupted()) {
        SocketConnection.this.shutdown(breakException, false);
      }
    }
  }

  /**
   * A thread dispatching V8 responses (to avoid locking the ReaderThread.)
   */
  private class ResponseDispatcherThread extends Thread {

    public ResponseDispatcherThread() {
      super("ResponseDispatcherThread");
    }

    @Override
    public void run() {
      MessageItem messageItem;
      try {
        while (true) {
          messageItem = inboundQueue.take();
          try {
            messageItem.report(listener);
          } catch (Exception e) {
            LOGGER.log(Level.SEVERE, "Exception in message listener", e);
          }
          if (messageItem.isEos()) {
            if (connectionLogger != null) {
              connectionLogger.handleEos();
            }
            break;
          }
        }
      } catch (InterruptedException e) {
        // terminate thread
      }
    }
  }

  /** The class logger. */
  private static final Logger LOGGER = Logger.getLogger(SocketConnection.class.getName());

  /** Lameduck shutdown delay in ms. */
  private static final int LAMEDUCK_DELAY_MS = 1000;

  /** The input stream buffer size. */
  private static final int INPUT_BUFFER_SIZE_BYTES = 65536;

  private static final NetListener NULL_LISTENER = new NetListener() {
    public void connectionClosed() {
    }

    public void eosReceived() {
    }

    public void messageReceived(Message message) {
    }
  };

  /** Whether the agent is currently attached to a remote browser. */
  private AtomicBoolean isAttached = new AtomicBoolean(false);

  /** The communication socket. */
  protected Socket socket;

  /** The socket reader. */
  protected BufferedReader reader;

  /** The socket writer. */
  protected BufferedWriter writer;

  private final ConnectionLogger connectionLogger;

  /** Handshaker used to establish connection. */
  private final Handshaker handshaker;

  /** The listener to report network events to. */
  protected volatile NetListener listener;

  /** The inbound message queue. */
  protected final BlockingQueue<MessageItem> inboundQueue = new LinkedBlockingQueue<MessageItem>();

  /** The outbound message queue. */
  protected final BlockingQueue<Message> outboundQueue = new LinkedBlockingQueue<Message>();

  /** The socket endpoint. */
  private final SocketAddress socketEndpoint;

  /** The thread that processes the outbound queue. */
  private WriterThread writerThread;

  /** The thread that processes the inbound queue. */
  private ReaderThread readerThread;

  /** Connection attempt timeout in ms. */
  private final int connectionTimeoutMs;

  public SocketConnection(SocketAddress endpoint, int connectionTimeoutMs,
      ConnectionLogger connectionLogger, Handshaker handshaker) {
    this.socketEndpoint = endpoint;
    this.connectionTimeoutMs = connectionTimeoutMs;
    this.connectionLogger = connectionLogger;
    this.handshaker = handshaker;
  }

  void attach() throws IOException {
    this.socket = new Socket();
    this.socket.connect(socketEndpoint, connectionTimeoutMs);
    Writer streamWriter = new OutputStreamWriter(socket.getOutputStream(), SOCKET_CHARSET);
    Reader streamReader = new InputStreamReader(socket.getInputStream(), SOCKET_CHARSET);

    if (connectionLogger != null) {
      streamWriter = connectionLogger.wrapWriter(streamWriter);
      streamReader = connectionLogger.wrapReader(streamReader);
      connectionLogger.setConnectionCloser(new ConnectionLogger.ConnectionCloser() {
        public void closeConnection() {
          close();
        }
      });
    }

    this.writer = new BufferedWriter(streamWriter);
    this.reader = new BufferedReader(streamReader, INPUT_BUFFER_SIZE_BYTES);
    isAttached.set(true);

    this.readerThread = new ReaderThread(reader, writer);
    // We do not start WriterThread until handshake is done (see ReaderThread)
    this.writerThread = null;
    readerThread.setDaemon(true);
    readerThread.start();
  }

  void detach(boolean lameduckMode) {
    shutdown(null, lameduckMode);
  }

  void sendMessage(Message message) {
    outboundQueue.add(message);
  }

  private boolean isAttached() {
    return isAttached.get();
  }

  /**
   * The method is synchronized so that it does not get called
   * from the {Reader,Writer}Thread when the underlying socket is
   * closed in another invocation of this method.
   */
  private void shutdown(Exception cause, boolean lameduckMode) {
    if (!isAttached.compareAndSet(true, false)) {
      // already shut down
      return;
    }
    LOGGER.log(Level.INFO, "Shutdown requested", cause);

    if (lameduckMode) {
      Thread terminationThread = new Thread("ServiceThreadTerminator") {
        @Override
        public void run() {
          interruptServiceThreads();
        }
      };
      terminationThread.setDaemon(true);
      terminationThread.start();
      try {
        terminationThread.join(LAMEDUCK_DELAY_MS);
      } catch (InterruptedException e) {
        // fall through
      }
    } else {
      interruptServiceThreads();
    }

    try {
      socket.shutdownInput();
    } catch (IOException e) {
      // ignore
    }
    try {
      socket.shutdownOutput();
    } catch (IOException e) {
      // ignore
    }

    try {
      socket.close();
    } catch (IOException e) {
      // ignore
    }
    listener.connectionClosed();
  }

  private void interruptServiceThreads() {
    interruptThread(writerThread);
    interruptThread(readerThread);
  }

  private void startWriterThread() {
    if (writerThread != null) {
      throw new IllegalStateException();
    }
    writerThread = new WriterThread(writer);
    writerThread.setDaemon(true);
    writerThread.start();
  }

  private ResponseDispatcherThread startResponseDispatcherThread() {
    ResponseDispatcherThread dispatcherThread;
    dispatcherThread = new ResponseDispatcherThread();
    dispatcherThread.setDaemon(true);
    dispatcherThread.start();
    return dispatcherThread;
  }

  private void interruptThread(Thread thread) {
    try {
      if (thread != null) {
        thread.interrupt();
      }
    } catch (SecurityException e) {
      // ignore
    }
  }

  public void close() {
    if (isAttached()) {
      detach(true);
    }
  }

  public boolean isConnected() {
    return isAttached();
  }

  public void send(Message message) {
    checkAttached();
    sendMessage(message);
  }

  public void setNetListener(NetListener netListener) {
    if (this.listener != null && netListener != this.listener) {
      throw new IllegalStateException("Cannot change NetListener");
    }
    this.listener = netListener != null
        ? netListener
        : NULL_LISTENER;
  }

  public void start() throws IOException {
    try {
      if (!isAttached()) {
        attach();
      }
    } catch (IOException e) {
      listener.connectionClosed();
      throw e;
    }
  }

  private void checkAttached() {
    if (!isAttached()) {
      throw new IllegalStateException("Connection not attached");
    }
  }
}