/*
 * Copyright 2013 gitblit.com.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.gitblit.fanout;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.text.MessageFormat;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A multi-threaded socket implementation of https://github.com/travisghansen/fanout
 *
 * This implementation creates a master acceptor thread which accepts incoming
 * fanout connections and then spawns a daemon thread for each accepted connection.
 * If there are 100 concurrent fanout connections, there are 101 threads.
 *
 * @author James Moger
 *
 */
public class FanoutSocketService extends FanoutService {

	private final static Logger logger = LoggerFactory.getLogger(FanoutSocketService.class);

	private volatile ServerSocket serviceSocket;

	public static void main(String[] args) throws Exception {
		FanoutSocketService pubsub = new FanoutSocketService(null, DEFAULT_PORT);
		pubsub.setStrictRequestTermination(false);
		pubsub.setAllowAllChannelAnnouncements(false);
		pubsub.start();
	}

	/**
	 * Create a multi-threaded fanout service.
	 *
	 * @param port
	 *            the port for running the fanout PubSub service
	 * @throws IOException
	 */
	public FanoutSocketService(int port) {
		this(null, port);
	}

	/**
	 * Create a multi-threaded fanout service.
	 *
	 * @param bindInterface
	 *            the ip address to bind for the service, may be null
	 * @param port
	 *            the port for running the fanout PubSub service
	 * @throws IOException
	 */
	public FanoutSocketService(String bindInterface, int port) {
		super(bindInterface, port, "Fanout socket service");
	}

	@Override
	protected boolean isConnected() {
		return serviceSocket != null;
	}

	@Override
	protected boolean connect() {
		if (serviceSocket == null) {
			try {
				serviceSocket = new ServerSocket();
				serviceSocket.setReuseAddress(true);
				serviceSocket.setSoTimeout(serviceTimeout);
				serviceSocket.bind(host == null ? new InetSocketAddress(port) : new InetSocketAddress(host, port));
				logger.info(MessageFormat.format("{0} is ready on {1}:{2,number,0}",
						name, host == null ? "0.0.0.0" : host, serviceSocket.getLocalPort()));
			} catch (IOException e) {
				logger.error(MessageFormat.format("failed to open {0} on {1}:{2,number,0}",
						name, host == null ? "0.0.0.0" : host, port), e);
				return false;
			}
		}
		return true;
	}

	@Override
	protected void disconnect() {
		try {
			if (serviceSocket != null) {
				logger.debug(MessageFormat.format("closing {0} server socket", name));
				serviceSocket.close();
				serviceSocket = null;
			}
		} catch (IOException e) {
			logger.error(MessageFormat.format("failed to disconnect {0}", name), e);
		}
	}

	/**
	 * This accepts incoming fanout connections and spawns connection threads.
	 */
	@Override
	protected void listen() throws IOException {
		try {
			Socket socket;
			socket = serviceSocket.accept();
			configureClientSocket(socket);

			FanoutSocketConnection connection = new FanoutSocketConnection(socket);

			if (addConnection(connection)) {
				// spawn connection daemon thread
				Thread connectionThread = new Thread(connection);
				connectionThread.setDaemon(true);
				connectionThread.setName("Fanout " + connection.id);
				connectionThread.start();
			} else {
				// synchronously close the connection and remove it
				removeConnection(connection);
				connection.closeConnection();
				connection = null;
			}
		} catch (SocketTimeoutException e) {
			// ignore accept timeout exceptions
		}
	}

	/**
	 * FanoutSocketConnection handles reading/writing messages from a remote fanout
	 * connection.
	 *
	 * @author James Moger
	 *
	 */
	class FanoutSocketConnection extends FanoutServiceConnection implements Runnable {
		Socket socket;

		FanoutSocketConnection(Socket socket) {
			super(socket);
			this.socket = socket;
		}

		/**
		 * Connection thread read/write method.
		 */
		@Override
		public void run() {
			try {
				StringBuilder sb = new StringBuilder();
				BufferedInputStream is = new BufferedInputStream(socket.getInputStream());
				byte[] buffer = new byte[FanoutConstants.BUFFER_LENGTH];
				int len = 0;
				while (true) {
					while (is.available() > 0) {
						len = is.read(buffer);
						for (int i = 0; i < len; i++) {
							byte b = buffer[i];
							if (b == 0xa || (!isStrictRequestTermination() && b == 0xd)) {
								String req = sb.toString();
								sb.setLength(0);
								if (req.length() > 0) {
									// ignore empty request strings
									processRequest(this, req);
								}
							} else {
								sb.append((char) b);
							}
						}
					}

					if (!isRunning.get()) {
						// service has stopped, terminate client connection
						break;
					} else {
						Thread.sleep(500);
					}
				}
			} catch (Throwable t) {
				if (t instanceof SocketException) {
					logger.error(MessageFormat.format("fanout connection {0}: {1}", id, t.getMessage()));
				} else if (t instanceof SocketTimeoutException) {
					logger.error(MessageFormat.format("fanout connection {0}: {1}", id, t.getMessage()));
				} else {
					logger.error(MessageFormat.format("exception while handling fanout connection {0}", id), t);
				}
			} finally {
				closeConnection();
			}

			logger.info(MessageFormat.format("thread for fanout connection {0} is finished", id));
		}

		@Override
		protected void reply(String content) throws IOException {
			// synchronously send reply
			logger.debug(MessageFormat.format("fanout reply to {0}: {1}", id, content));
			OutputStream os = socket.getOutputStream();
			byte [] bytes = content.getBytes(FanoutConstants.CHARSET);
			os.write(bytes);
			if (bytes[bytes.length - 1] != 0xa) {
				os.write(0xa);
			}
			os.flush();
		}

		protected void closeConnection() {
			// close the connection socket
			try {
				socket.close();
			} catch (IOException e) {
			}
			socket = null;

			// remove this connection from the service
			removeConnection(this);
		}
	}
}