/*
    * Copyright (C) 2013 4th Line GmbH, Switzerland
    *
    * The contents of this file are subject to the terms of either the GNU
    * Lesser General Public License Version 2 or later ("LGPL") or the
    * Common Development and Distribution License Version 1 or later
    * ("CDDL") (collectively, the "License"). You may not use this file
    * except in compliance with the License. See LICENSE.txt for more
    * information.
   *
   * 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.
   */
  
  package org.fourthline.cling.transport;
  
  import org.fourthline.cling.UpnpServiceConfiguration;
  import org.fourthline.cling.model.NetworkAddress;
  import org.fourthline.cling.model.message.IncomingDatagramMessage;
  import org.fourthline.cling.model.message.OutgoingDatagramMessage;
  import org.fourthline.cling.model.message.StreamRequestMessage;
  import org.fourthline.cling.model.message.StreamResponseMessage;
  import org.fourthline.cling.protocol.ProtocolCreationException;
  import org.fourthline.cling.protocol.ProtocolFactory;
  import org.fourthline.cling.protocol.ReceivingAsync;
  import org.fourthline.cling.transport.spi.DatagramIO;
  import org.fourthline.cling.transport.spi.InitializationException;
  import org.fourthline.cling.transport.spi.MulticastReceiver;
  import org.fourthline.cling.transport.spi.NetworkAddressFactory;
  import org.fourthline.cling.transport.spi.NoNetworkException;
  import org.fourthline.cling.transport.spi.StreamClient;
  import org.fourthline.cling.transport.spi.StreamServer;
  import org.fourthline.cling.transport.spi.UpnpStream;
  import org.seamless.util.Exceptions;
  
  import javax.enterprise.context.ApplicationScoped;
  import javax.enterprise.event.Observes;
  import javax.enterprise.inject.Default;
  import javax.inject.Inject;
  import java.net.BindException;
  import java.net.DatagramPacket;
  import java.net.InetAddress;
  import java.net.NetworkInterface;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  /**
   * Default implementation of network message router.
   * <p>
   * Initializes and starts listening for data on the network when enabled.
   * </p>
   *
   * @author Christian Bauer
   */
  @ApplicationScoped
  public class RouterImpl implements Router {
  
      private static Logger log = Logger.getLogger(Router.class.getName());
  
      protected UpnpServiceConfiguration configuration;
      protected ProtocolFactory protocolFactory;
  
      protected volatile boolean enabled;
      protected ReentrantReadWriteLock routerLock = new ReentrantReadWriteLock(true);
      protected Lock readLock = routerLock.readLock();
      protected Lock writeLock = routerLock.writeLock();
  
      // These are created/destroyed when the router is enabled/disabled
      protected NetworkAddressFactory networkAddressFactory;
      protected StreamClient streamClient;
      protected final Map<NetworkInterface, MulticastReceiver> multicastReceivers = new HashMap<>();
      protected final Map<InetAddress, DatagramIO> datagramIOs = new HashMap<>();
      protected final Map<InetAddress, StreamServer> streamServers = new HashMap<>();
  
      protected RouterImpl() {
      }
  
      /**
       * @param configuration   The configuration used by this router.
       * @param protocolFactory The protocol factory used by this router.
       */
      @Inject
      public RouterImpl(UpnpServiceConfiguration configuration, ProtocolFactory protocolFactory) {
          log.info("Creating Router: " + getClass().getName());
          this.configuration = configuration;
          this.protocolFactory = protocolFactory;
      }
  
      public boolean enable(@Observes @Default EnableRouter event) throws RouterException {
         return enable();
     }
 
     public boolean disable(@Observes @Default DisableRouter event) throws RouterException {
         return disable();
     }
 
     public UpnpServiceConfiguration getConfiguration() {
         return configuration;
     }
 
     public ProtocolFactory getProtocolFactory() {
         return protocolFactory;
     }
 
     /**
      * Initializes listening services: First an instance of {@link org.fourthline.cling.transport.spi.MulticastReceiver}
      * is bound to each network interface. Then an instance of {@link org.fourthline.cling.transport.spi.DatagramIO} and
      * {@link org.fourthline.cling.transport.spi.StreamServer} is bound to each bind address returned by the network
      * address factory, respectively. There is only one instance of
      * {@link org.fourthline.cling.transport.spi.StreamClient} created and managed by this router.
      */
     @Override
     public boolean enable() throws RouterException {
         lock(writeLock);
         try {
             if (!enabled) {
                 try {
                     log.fine("Starting networking services...");
                     networkAddressFactory = getConfiguration().createNetworkAddressFactory();
 
                     startInterfaceBasedTransports(networkAddressFactory.getNetworkInterfaces());
                     startAddressBasedTransports(networkAddressFactory.getBindAddresses());
 
                     // The transports possibly removed some unusable network interfaces/addresses
                     if (!networkAddressFactory.hasUsableNetwork()) {
                         throw new NoNetworkException(
                             "No usable network interface and/or addresses available, check the log for errors."
                         );
                     }
 
                     // Start the HTTP client last, we don't even have to try if there is no network
                     streamClient = getConfiguration().createStreamClient();
 
                     enabled = true;
                     return true;
                 } catch (InitializationException ex) {
                     handleStartFailure(ex);
                 }
             }
             return false;
         } finally {
             unlock(writeLock);
         }
     }
 
     @Override
     public boolean disable() throws RouterException {
         lock(writeLock);
         try {
             if (enabled) {
                 log.fine("Disabling network services...");
 
                 if (streamClient != null) {
                     log.fine("Stopping stream client connection management/pool");
                     streamClient.stop();
                     streamClient = null;
                 }
 
                 for (Map.Entry<InetAddress, StreamServer> entry : streamServers.entrySet()) {
                     log.fine("Stopping stream server on address: " + entry.getKey());
                     entry.getValue().stop();
                 }
                 streamServers.clear();
 
                 for (Map.Entry<NetworkInterface, MulticastReceiver> entry : multicastReceivers.entrySet()) {
                     log.fine("Stopping multicast receiver on interface: " + entry.getKey().getDisplayName());
                     entry.getValue().stop();
                 }
                 multicastReceivers.clear();
 
                 for (Map.Entry<InetAddress, DatagramIO> entry : datagramIOs.entrySet()) {
                     log.fine("Stopping datagram I/O on address: " + entry.getKey());
                     entry.getValue().stop();
                 }
                 datagramIOs.clear();
 
                 networkAddressFactory = null;
                 enabled = false;
                 return true;
             }
             return false;
         } finally {
             unlock(writeLock);
         }
     }
 
     @Override
     public void shutdown() throws RouterException {
         disable();
     }
 
     @Override
     public boolean isEnabled() {
         return enabled;
     }
 
     @Override
     public void handleStartFailure(InitializationException ex) throws InitializationException {
         if (ex instanceof NoNetworkException) {
             log.info("Unable to initialize network router, no network found.");
         } else {
             log.severe("Unable to initialize network router: " + ex);
             log.severe("Cause: " + Exceptions.unwrap(ex));
         }
     }
 
     public List<NetworkAddress> getActiveStreamServers(InetAddress preferredAddress) throws RouterException {
         lock(readLock);
         try {
             if (enabled && streamServers.size() > 0) {
                 List<NetworkAddress> streamServerAddresses = new ArrayList<>();
 
                 StreamServer preferredServer;
                 if (preferredAddress != null &&
                     (preferredServer = streamServers.get(preferredAddress)) != null) {
                     streamServerAddresses.add(
                         new NetworkAddress(
                             preferredAddress,
                             preferredServer.getPort(),
                             networkAddressFactory.getHardwareAddress(preferredAddress)
 
                         )
                     );
                     return streamServerAddresses;
                 }
 
                 for (Map.Entry<InetAddress, StreamServer> entry : streamServers.entrySet()) {
                     byte[] hardwareAddress = networkAddressFactory.getHardwareAddress(entry.getKey());
                     streamServerAddresses.add(
                         new NetworkAddress(entry.getKey(), entry.getValue().getPort(), hardwareAddress)
                     );
                 }
                 return streamServerAddresses;
             } else {
                 return Collections.EMPTY_LIST;
             }
         } finally {
             unlock(readLock);
         }
     }
 
     /**
      * Obtains the asynchronous protocol {@code Executor} and runs the protocol created
      * by the {@link org.fourthline.cling.protocol.ProtocolFactory} for the given message.
      * <p>
      * If the factory doesn't create a protocol, the message is dropped immediately without
      * creating another thread or consuming further resources. This means we can filter the
      * datagrams in the protocol factory and e.g. completely disable discovery or only
      * allow notification message from some known services we'd like to work with.
      * </p>
      *
      * @param msg The received datagram message.
      */
     public void received(IncomingDatagramMessage msg) {
         if (!enabled) {
             log.fine("Router disabled, ignoring incoming message: " + msg);
             return;
         }
         try {
             ReceivingAsync protocol = getProtocolFactory().createReceivingAsync(msg);
             if (protocol == null) {
                 if (log.isLoggable(Level.FINEST))
                     log.finest("No protocol, ignoring received message: " + msg);
                 return;
             }
             if (log.isLoggable(Level.FINE))
                 log.fine("Received asynchronous message: " + msg);
             getConfiguration().getAsyncProtocolExecutor().execute(protocol);
         } catch (ProtocolCreationException ex) {
             log.warning("Handling received datagram failed - " + Exceptions.unwrap(ex).toString());
         }
     }
 
     /**
      * Obtains the synchronous protocol {@code Executor} and runs the
      * {@link org.fourthline.cling.transport.spi.UpnpStream} directly.
      *
      * @param stream The received {@link org.fourthline.cling.transport.spi.UpnpStream}.
      */
     public void received(UpnpStream stream) {
         if (!enabled) {
             log.fine("Router disabled, ignoring incoming: " + stream);
             return;
         }
         log.fine("Received synchronous stream: " + stream);
         getConfiguration().getSyncProtocolExecutorService().execute(stream);
     }
 
     /**
      * Sends the UDP datagram on all bound {@link org.fourthline.cling.transport.spi.DatagramIO}s.
      *
      * @param msg The UDP datagram message to send.
      */
     public void send(OutgoingDatagramMessage msg) throws RouterException {
         lock(readLock);
         try {
             if (enabled) {
                 for (DatagramIO datagramIO : datagramIOs.values()) {
                     datagramIO.send(msg);
                 }
             } else {
                 log.fine("Router disabled, not sending datagram: " + msg);
             }
         } finally {
             unlock(readLock);
         }
     }
 
     /**
      * Sends the TCP stream request with the {@link org.fourthline.cling.transport.spi.StreamClient}.
      *
      * @param msg The TCP (HTTP) stream message to send.
      * @return The return value of the {@link org.fourthline.cling.transport.spi.StreamClient#sendRequest(StreamRequestMessage)}
      *         method or <code>null</code> if no <code>StreamClient</code> is available.
      */
     public StreamResponseMessage send(StreamRequestMessage msg) throws RouterException {
         lock(readLock);
         try {
             if (enabled) {
                 if (streamClient == null) {
                     log.fine("No StreamClient available, not sending: " + msg);
                     return null;
                 }
                 log.fine("Sending via TCP unicast stream: " + msg);
                 try {
                     return streamClient.sendRequest(msg);
                 } catch (InterruptedException ex) {
                     throw new RouterException("Sending stream request was interrupted", ex);
                 }
             } else {
                 log.fine("Router disabled, not sending stream request: " + msg);
                 return null;
             }
         } finally {
             unlock(readLock);
         }
     }
 
     /**
      * Sends the given bytes as a broadcast on all bound {@link org.fourthline.cling.transport.spi.DatagramIO}s,
      * using source port 9.
      * <p>
      * TODO: Support source port parameter
      * </p>
      *
      * @param bytes The byte payload of the UDP datagram.
      */
     public void broadcast(byte[] bytes) throws RouterException {
         lock(readLock);
         try {
             if (enabled) {
                 for (Map.Entry<InetAddress, DatagramIO> entry : datagramIOs.entrySet()) {
                     InetAddress broadcast = networkAddressFactory.getBroadcastAddress(entry.getKey());
                     if (broadcast != null) {
                         log.fine("Sending UDP datagram to broadcast address: " + broadcast.getHostAddress());
                         DatagramPacket packet = new DatagramPacket(bytes, bytes.length, broadcast, 9);
                         entry.getValue().send(packet);
                     }
                 }
             } else {
                 log.fine("Router disabled, not broadcasting bytes: " + bytes.length);
             }
         } finally {
             unlock(readLock);
         }
     }
 
     protected void startInterfaceBasedTransports(Iterator<NetworkInterface> interfaces) throws InitializationException {
         while (interfaces.hasNext()) {
             NetworkInterface networkInterface = interfaces.next();
 
             // We only have the MulticastReceiver as an interface-based transport
             MulticastReceiver multicastReceiver = getConfiguration().createMulticastReceiver(networkAddressFactory);
             if (multicastReceiver == null) {
                 log.info("Configuration did not create a MulticastReceiver for: " + networkInterface);
             } else {
                 try {
                     if (log.isLoggable(Level.FINE))
                         log.fine("Init multicast receiver on interface: " + networkInterface.getDisplayName());
                     multicastReceiver.init(
                         networkInterface,
                         this,
                         networkAddressFactory,
                         getConfiguration().getDatagramProcessor()
                     );
 
                     multicastReceivers.put(networkInterface, multicastReceiver);
                 } catch (InitializationException ex) {
                     /* TODO: What are some recoverable exceptions for this?
                     log.warning(
                         "Ignoring network interface '"
                             + networkInterface.getDisplayName()
                             + "' init failure of MulticastReceiver: " + ex.toString());
                     if (log.isLoggable(Level.FINE))
                         log.log(Level.FINE, "Initialization exception root cause", Exceptions.unwrap(ex));
                     log.warning("Removing unusable interface " + interface);
                     it.remove();
                     continue; // Don't need to try anything else on this interface
                     */
                     throw ex;
                 }
             }
         }
 
         for (Map.Entry<NetworkInterface, MulticastReceiver> entry : multicastReceivers.entrySet()) {
             if (log.isLoggable(Level.FINE))
                 log.fine("Starting multicast receiver on interface: " + entry.getKey().getDisplayName());
             getConfiguration().getMulticastReceiverExecutor().execute(entry.getValue());
         }
     }
 
     protected void startAddressBasedTransports(Iterator<InetAddress> addresses) throws InitializationException {
         while (addresses.hasNext()) {
             InetAddress address = addresses.next();
 
             // HTTP servers
             StreamServer streamServer = getConfiguration().createStreamServer(networkAddressFactory);
             if (streamServer == null) {
                 log.info("Configuration did not create a StreamServer for: " + address);
             } else {
                 try {
                     if (log.isLoggable(Level.FINE))
                         log.fine("Init stream server on address: " + address);
                     streamServer.init(address, this);
                     streamServers.put(address, streamServer);
                 } catch (InitializationException ex) {
                     // Try to recover
                     Throwable cause = Exceptions.unwrap(ex);
                     if (cause instanceof BindException) {
                         log.warning("Failed to init StreamServer: " + cause);
                         if (log.isLoggable(Level.FINE))
                             log.log(Level.FINE, "Initialization exception root cause", cause);
                         log.warning("Removing unusable address: " + address);
                         addresses.remove();
                         continue; // Don't try anything else with this address
                     }
                     throw ex;
                 }
             }
 
             // Datagram I/O
             DatagramIO datagramIO = getConfiguration().createDatagramIO(networkAddressFactory);
             if (datagramIO == null) {
                 log.info("Configuration did not create a StreamServer for: " + address);
             } else {
                 try {
                     if (log.isLoggable(Level.FINE))
                         log.fine("Init datagram I/O on address: " + address);
                     datagramIO.init(address, this, getConfiguration().getDatagramProcessor());
                     datagramIOs.put(address, datagramIO);
                 } catch (InitializationException ex) {
                     /* TODO: What are some recoverable exceptions for this?
                     Throwable cause = Exceptions.unwrap(ex);
                     if (cause instanceof BindException) {
                         log.warning("Failed to init datagram I/O: " + cause);
                         if (log.isLoggable(Level.FINE))
                             log.log(Level.FINE, "Initialization exception root cause", cause);
                         log.warning("Removing unusable address: " + address);
                         addresses.remove();
                         continue; // Don't try anything else with this address
                     }
                     */
                     throw ex;
                 }
             }
         }
 
         for (Map.Entry<InetAddress, StreamServer> entry : streamServers.entrySet()) {
             if (log.isLoggable(Level.FINE))
                 log.fine("Starting stream server on address: " + entry.getKey());
             getConfiguration().getStreamServerExecutorService().execute(entry.getValue());
         }
 
         for (Map.Entry<InetAddress, DatagramIO> entry : datagramIOs.entrySet()) {
             if (log.isLoggable(Level.FINE))
                 log.fine("Starting datagram I/O on address: " + entry.getKey());
             getConfiguration().getDatagramIOExecutor().execute(entry.getValue());
         }
     }
 
     protected void lock(Lock lock, int timeoutMilliseconds) throws RouterException {
         try {
             log.finest("Trying to obtain lock with timeout milliseconds '" + timeoutMilliseconds + "': " + lock.getClass().getSimpleName());
             if (lock.tryLock(timeoutMilliseconds, TimeUnit.MILLISECONDS)) {
                 log.finest("Acquired router lock: " + lock.getClass().getSimpleName());
             } else {
                 throw new RouterException(
                     "Router wasn't available exclusively after waiting " + timeoutMilliseconds + "ms, lock failed: "
                         + lock.getClass().getSimpleName()
                 );
             }
         } catch (InterruptedException ex) {
             throw new RouterException(
                 "Interruption while waiting for exclusive access: " + lock.getClass().getSimpleName(), ex
             );
         }
     }
 
     protected void lock(Lock lock) throws RouterException {
         lock(lock, getLockTimeoutMillis());
     }
 
     protected void unlock(Lock lock) {
         log.finest("Releasing router lock: " + lock.getClass().getSimpleName());
         lock.unlock();
     }
 
     /**
      * @return Defaults to 6 seconds, should be longer than it takes the router to be enabled/disabled.
      */
     protected int getLockTimeoutMillis() {
         return 6000;
     }
 
 }