/*

  *   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());

 			}

 		}

 	}

 }