/*

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

      }

    }

  }

}