/*
    * 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 org.apache.log4j.plugins.Pauseable;
  import org.apache.log4j.plugins.Receiver;
  import org.apache.log4j.spi.Decoder;
  import org.apache.log4j.spi.LoggingEvent;
  
  import java.io.IOException;
  import java.net.DatagramPacket;
  import java.net.DatagramSocket;
  import java.net.SocketException;
  import java.util.ArrayList;
  import java.util.List;
  
  
  /**
   * 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 | InstantiationException iae) {
             getLogger().warn("Could not construct decoder", iae);
         }
 
         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 final List<String> 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<String> 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) {
 
                     for (Object aList2 : list2) {
                         String data = (String) aList2;
                         List<LoggingEvent> v = decoderImpl.decodeEvents(data);
 
                         if (v != null) {
 
                             for (Object aV : v) {
                                 if (!isPaused()) {
                                     doPost((LoggingEvent) aV);
                                 }
                             }
                         }
                     }
 
                     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.");
         }
     }
 }