/*
    * Licensed to the Apache Software Foundation (ASF) under one or more
    * contributor license agreements.  See the NOTICE file distributed with
    * this work for additional information regarding copyright ownership.
    * The ASF licenses this file to You 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 org.apache.log4j.net;
  
  import java.io.IOException;
  import java.net.DatagramPacket;
  import java.net.DatagramSocket;
  import java.net.SocketException;
  import java.util.ArrayList;
  import java.util.Iterator;
  import java.util.List;
  
  import org.apache.log4j.plugins.Pauseable;
  import org.apache.log4j.plugins.Receiver;
  import org.apache.log4j.spi.Decoder;
  import org.apache.log4j.spi.LoggingEvent;
  
  
  /**
   *  Receive LoggingEvents encoded with an XMLLayout, convert the XML data to a
   *  LoggingEvent and post the LoggingEvent.
   *
   *  @author Scott Deboy <sdeboy@apache.org>
   *
   */
  public class UDPReceiver extends Receiver implements PortBased, Pauseable {
    private static final int PACKET_LENGTH = 16384;
    private UDPReceiverThread receiverThread;
    private String encoding;
  
    //default to log4j xml decoder
    private String decoder = "org.apache.log4j.xml.XMLDecoder";
    private Decoder decoderImpl;
    protected boolean paused;
    private transient boolean closed = false;
    private int port;
    private DatagramSocket socket;
    UDPHandlerThread handlerThread;
    private boolean advertiseViaMulticastDNS;
    private ZeroConfSupport zeroConf;
  
    /**
     * The MulticastDNS zone advertised by a UDPReceiver
     */
    public static final String ZONE = "_log4j_xml_udp_receiver.local.";
  
  
      public int getPort() {
      return port;
    }
  
    public void setPort(int port) {
      this.port = port;
    }
  
    /**
     * The <b>Encoding</b> option specifies how the bytes are encoded.  If this 
     * option is not specified, the system encoding will be used.
     * */
    public void setEncoding(String encoding) {
      this.encoding = encoding;
    }
  
    /**
     * Returns value of the <b>Encoding</b> option.
     */
    public String getEncoding() {
      return encoding;
    }
  
    public String getDecoder() {
      return decoder;
    }
  
    public void setDecoder(String decoder) {
      this.decoder = decoder;
    }
  
    public boolean isPaused() {
      return paused;
    }
  
    public void setPaused(boolean b) {
      paused = b;
   }
 
   public void setAdvertiseViaMulticastDNS(boolean advertiseViaMulticastDNS) {
     this.advertiseViaMulticastDNS = advertiseViaMulticastDNS;
   }
 
   public boolean isAdvertiseViaMulticastDNS() {
     return advertiseViaMulticastDNS;
   }
 
   public synchronized void shutdown() {
     if(closed == true) {
       return;
     }
     closed = true;
     active = false;
     // Closing the datagram socket will unblock the UDPReceiverThread if it is
     // was waiting to receive data from the socket.
     if (socket != null) {
     	socket.close();
     }
 
     if (advertiseViaMulticastDNS) {
       zeroConf.unadvertise();
     }
       
     try {
       if(handlerThread != null) {
       	handlerThread.close();
         handlerThread.join();
       }
       if(receiverThread != null) {
         receiverThread.join();
       }
     } catch(InterruptedException ie) {
     }
   }
 
   /**
     Returns true if this receiver is active. */
 //  public synchronized boolean isActive() {
 //    return isActive;
 //}
 
   public void activateOptions() {
     try {
       Class c = Class.forName(decoder);
       Object o = c.newInstance();
 
       if (o instanceof Decoder) {
         this.decoderImpl = (Decoder) o;
       }
     } catch (ClassNotFoundException cnfe) {
       getLogger().warn("Unable to find decoder", cnfe);
     } catch (IllegalAccessException iae) {
       getLogger().warn("Could not construct decoder", iae);
     } catch (InstantiationException ie) {
       getLogger().warn("Could not construct decoder", ie);
     }
 
     try {
       socket = new DatagramSocket(port);
       receiverThread = new UDPReceiverThread();
       receiverThread.start();
       handlerThread = new UDPHandlerThread();
       handlerThread.start();
       if (advertiseViaMulticastDNS) {
         zeroConf = new ZeroConfSupport(ZONE, port, getName());
         zeroConf.advertise();
       }
       active = true;
     } catch (IOException ioe) {
       ioe.printStackTrace();
     }
   }
 
   class UDPHandlerThread extends Thread {
     private List list = new ArrayList();
 
     public UDPHandlerThread() {
       setDaemon(true);
     }
 
     public void append(String data) {
       synchronized (list) {
         list.add(data);
         list.notify();
       }
     }
 
     /**
      * Allow the UDPHandlerThread to wakeup and exit gracefully.
      */
     void close() {
       synchronized(list) {
       	list.notify();
       }
     }
 
     public void run() {
       ArrayList list2 = new ArrayList();
 
       while (!UDPReceiver.this.closed) {
         synchronized (list) {
           try {
             while (!UDPReceiver.this.closed && list.size() == 0) {
               list.wait(300);
             }
 
             if (list.size() > 0) {
               list2.addAll(list);
               list.clear();
             }
           } catch (InterruptedException ie) {
           }
         }
 
         if (list2.size() > 0) {
           Iterator iter = list2.iterator();
 
           while (iter.hasNext()) {
             String data = (String) iter.next();
             List v = decoderImpl.decodeEvents(data);
 
             if (v != null) {
               Iterator eventIter = v.iterator();
 
               while (eventIter.hasNext()) {
                 if (!isPaused()) {
                   doPost((LoggingEvent) eventIter.next());
                 }
               }
             }
           }
 
           list2.clear();
         } else {
           try {
             synchronized (this) {
               wait(1000);
             }
           } catch (InterruptedException ie) {
           }
         }
       } // while
       getLogger().debug(UDPReceiver.this.getName()+ "'s handler thread is exiting");
     } // run
   } // UDPHandlerThread
 
   class UDPReceiverThread extends Thread {
     public UDPReceiverThread() {
       setDaemon(true);
     }
     
     public void run() {
       byte[] b = new byte[PACKET_LENGTH];
       DatagramPacket p = new DatagramPacket(b, b.length);
 
       while (!UDPReceiver.this.closed) {
         try {
           socket.receive(p);
           
           //this string constructor which accepts a charset throws an exception if it is 
           //null
           if (encoding == null) {
             handlerThread.append(
               new String(p.getData(), 0, p.getLength()));
           } else {
             handlerThread.append(
               new String(p.getData(), 0, p.getLength(), encoding));
           }
         } catch (SocketException se) {
           //disconnected
         } catch (IOException ioe) {
           ioe.printStackTrace();
         }
       }
 
       //LogLog.debug(UDPReceiver.this.getName() + "'s thread is ending.");
     }
   }
 }