/*

 *   SONEWS News Server

 *   see AUTHORS for the list of contributors

 *

 *   This program is free software: you can redistribute it and/or modify

 *   it under the terms of the GNU General Public License as published by

 *   the Free Software Foundation, either version 3 of the License, or

 *   (at your option) any later version.

 *

 *   This program is distributed in the hope that it will be useful,

 *   but WITHOUT ANY WARRANTY; without even the implied warranty of

 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

 *   GNU General Public License for more details.

 *

 *   You should have received a copy of the GNU General Public License

 *   along with this program.  If not, see <http://www.gnu.org/licenses/>.

 */

package org.sonews.feed;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintWriter;

import java.net.Socket;

import java.net.SocketException;

import java.net.UnknownHostException;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.HashSet;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.logging.Level;

import org.sonews.config.Config;

import org.sonews.daemon.AbstractDaemon;

import org.sonews.util.Log;

import org.sonews.storage.StorageBackendException;

import org.sonews.storage.StorageManager;

import org.sonews.util.Stats;

import org.sonews.util.io.ArticleReader;

import org.sonews.util.io.ArticleWriter;

/**

 * The PullFeeder class regularily checks another Newsserver for new

 * messages.

 * @author Christian Lins

 * @since sonews/0.5.0

 */

class PullFeeder extends AbstractDaemon

{

	private Map<Subscription, Integer> highMarks = new HashMap<Subscription, Integer>();

	private BufferedReader in;

	private PrintWriter out;

	private Set<Subscription> subscriptions = new HashSet<Subscription>();

	private void addSubscription(final Subscription sub)

	{

		subscriptions.add(sub);

		if (!highMarks.containsKey(sub)) {

			// Set a initial highMark

			this.highMarks.put(sub, 0);

		}

	}

	/**

	 * Changes to the given group and returns its high mark.

	 * @param groupName

	 * @return

	 */

	private int changeGroup(String groupName)

		throws IOException

	{

		this.out.print("GROUP " + groupName + "\r\n");

		this.out.flush();

		String line = this.in.readLine();

		if (line.startsWith("211 ")) {

			int highmark = Integer.parseInt(line.split(" ")[3]);

			return highmark;

		} else {

			throw new IOException("GROUP " + groupName + " returned: " + line);

		}

	}

	private void connectTo(final String host, final int port)

		throws IOException, UnknownHostException

	{

		Socket socket = new Socket(host, port);

		this.out = new PrintWriter(socket.getOutputStream());

		this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

		String line = in.readLine();

		if (!(line.charAt(0) == '2')) // Could be 200 or 2xx if posting is not allowed

		{

			throw new IOException(line);

		}

		// Send MODE READER to peer, some newsservers are friendlier then

		this.out.println("MODE READER\r\n");

		this.out.flush();

		line = this.in.readLine();

	}

	private void disconnect()

		throws IOException

	{

		this.out.print("QUIT\r\n");

		this.out.flush();

		this.out.close();

		this.in.close();

		this.out = null;

		this.in = null;

	}

	/**

	 * Uses the OVER or XOVER command to get a list of message overviews that

	 * may be unknown to this feeder and are about to be peered.

	 * @param start

	 * @param end

	 * @return A list of message ids with potentially interesting messages.

	 */

	private List<String> over(int start, int end)

		throws IOException

	{

		this.out.print("OVER " + start + "-" + end + "\r\n");

		this.out.flush();

		String line = this.in.readLine();

		if (line.startsWith("500 ")) // OVER not supported

		{

			this.out.print("XOVER " + start + "-" + end + "\r\n");

			this.out.flush();

			line = this.in.readLine();

		}

		if (line.startsWith("224 ")) {

			List<String> messages = new ArrayList<String>();

			line = this.in.readLine();

			while (!".".equals(line)) {

				String mid = line.split("\t")[4]; // 5th should be the Message-ID

				messages.add(mid);

				line = this.in.readLine();

			}

			return messages;

		} else {

			throw new IOException("Server return for OVER/XOVER: " + line);

		}

	}

	@Override

	public void run()

	{

		while (isRunning()) {

			int pullInterval = 1000

				* Config.inst().get(Config.FEED_PULLINTERVAL, 3600);

			String host = "localhost";

			int port = 119;

			Log.get().info("Start PullFeeder run...");

			try {

				this.subscriptions.clear();

				List<Subscription> subsPull = StorageManager.current().getSubscriptions(FeedManager.TYPE_PULL);

				for (Subscription sub : subsPull) {

					addSubscription(sub);

				}

			} catch (StorageBackendException ex) {

				Log.get().log(Level.SEVERE, host, ex);

			}

			try {

				for (Subscription sub : this.subscriptions) {

					host = sub.getHost();

					port = sub.getPort();

					try {

						Log.get().info("Feeding " + sub.getGroup() + " from " + sub.getHost());

						try {

							connectTo(host, port);

						} catch (SocketException ex) {

							Log.get().info("Skipping " + sub.getHost() + ": " + ex);

							continue;

						}

						int oldMark = this.highMarks.get(sub);

						int newMark = changeGroup(sub.getGroup());

						if (oldMark != newMark) {

							List<String> messageIDs = over(oldMark, newMark);

							for (String messageID : messageIDs) {

								if (!StorageManager.current().isArticleExisting(messageID)) {

									try {

										// Post the message via common socket connection

										ArticleReader aread =

											new ArticleReader(sub.getHost(), sub.getPort(), messageID);

										byte[] abuf = aread.getArticleData();

										if (abuf == null) {

											Log.get().warning("Could not feed " + messageID

												+ " from " + sub.getHost());

										} else {

											Log.get().info("Feeding " + messageID);

											ArticleWriter awrite = new ArticleWriter(

												"localhost", Config.inst().get(Config.PORT, 119));

											awrite.writeArticle(abuf);

											awrite.close();

										}

										Stats.getInstance().mailFeeded(sub.getGroup());

									} catch (IOException ex) {

										// There may be a temporary network failure

										ex.printStackTrace();

										Log.get().warning("Skipping mail " + messageID + " due to exception.");

									}

								}

							} // for(;;)

							this.highMarks.put(sub, newMark);

						}

						disconnect();

					} catch (StorageBackendException ex) {

						ex.printStackTrace();

					} catch (IOException ex) {

						ex.printStackTrace();

						Log.get().severe("PullFeeder run stopped due to exception.");

					}

				} // for(Subscription sub : subscriptions)

				Log.get().info("PullFeeder run ended. Waiting " + pullInterval / 1000 + "s");

				Thread.sleep(pullInterval);

			} catch (InterruptedException ex) {

				Log.get().warning(ex.getMessage());

			}

		}

	}

}