/*
 * 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.flume.source;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.MultiportSyslogTCPSource.LineSplitter;
import org.apache.flume.source.MultiportSyslogTCPSource.MultiportSyslogHandler;
import org.apache.flume.source.MultiportSyslogTCPSource.ParsedBuffer;
import org.apache.flume.source.MultiportSyslogTCPSource.ThreadSafeDecoder;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.DefaultIoSessionDataStructureFactory;
import org.apache.mina.transport.socket.nio.NioSession;
import org.joda.time.DateTime;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.mockito.Mockito.*;

public class TestMultiportSyslogTCPSource {

  private static final Logger logger =
      LoggerFactory.getLogger(TestMultiportSyslogTCPSource.class);

  private static final int BASE_TEST_SYSLOG_PORT = 14455;
  private final DateTime time = new DateTime();
  private final String stamp1 = time.toString();
  private final String host1 = "localhost.localdomain";
  private final String data1 = "proc1 - some msg";

  /**
   * Helper function to generate a syslog message.
   * @param counter
   * @return 
   */
  private byte[] getEvent(int counter) {
    // timestamp with 'Z' appended, translates to UTC
    String msg1 = "<10>" + stamp1 + " " + host1 + " " + data1 + " "
            + String.valueOf(counter) + "\n";
    return msg1.getBytes();
  }

  /**
   * Basic test to exercise multiple-port parsing.
   */
  @Test
  public void testMultiplePorts() throws IOException, ParseException {
    MultiportSyslogTCPSource source = new MultiportSyslogTCPSource();
    Channel channel = new MemoryChannel();

    Context channelContext = new Context();
    channelContext.put("capacity", String.valueOf(2000));
    channelContext.put("transactionCapacity", String.valueOf(2000));
    Configurables.configure(channel, channelContext);

    List<Channel> channels = Lists.newArrayList();
    channels.add(channel);

    ChannelSelector rcs = new ReplicatingChannelSelector();
    rcs.setChannels(channels);

    source.setChannelProcessor(new ChannelProcessor(rcs));
    Context context = new Context();
    StringBuilder ports = new StringBuilder();
    for (int i = 0; i < 1000; i++) {
      ports.append(String.valueOf(BASE_TEST_SYSLOG_PORT + i)).append(" ");
    }
    context.put(SyslogSourceConfigurationConstants.CONFIG_PORTS,
        ports.toString().trim());
    source.configure(context);
    source.start();

    Socket syslogSocket;
    for (int i = 0; i < 1000 ; i++) {
      syslogSocket = new Socket(
              InetAddress.getLocalHost(), BASE_TEST_SYSLOG_PORT + i);
      syslogSocket.getOutputStream().write(getEvent(i));
      syslogSocket.close();
    }

    List<Event> channelEvents = new ArrayList<Event>();
    Transaction txn = channel.getTransaction();
    txn.begin();
    for (int i = 0; i < 1000; i++) {
      Event e = channel.take();
      if (e == null) {
        throw new NullPointerException("Event is null");
      }
      channelEvents.add(e);
    }
    try {
      txn.commit();
    } catch (Throwable t) {
      txn.rollback();
    } finally {
      txn.close();
    }
    //Since events can arrive out of order, search for each event in the array
    for (int i = 0; i < 1000 ; i++) {
      Iterator<Event> iter = channelEvents.iterator();
      while (iter.hasNext()) {
        Event e = iter.next();
        Map<String, String> headers = e.getHeaders();
        // rely on port to figure out which event it is
        Integer port = null;
        if (headers.containsKey(
            SyslogSourceConfigurationConstants.DEFAULT_PORT_HEADER)) {
          port = Integer.parseInt(headers.get(
                SyslogSourceConfigurationConstants.DEFAULT_PORT_HEADER));
        }
        iter.remove();

        Assert.assertEquals("Timestamps must match",
            String.valueOf(time.getMillis()), headers.get("timestamp"));

        String host2 = headers.get("host");
        Assert.assertEquals(host1, host2);

        if (port != null) {
          int num = port - BASE_TEST_SYSLOG_PORT;
          Assert.assertEquals(data1 + " " + String.valueOf(num),
              new String(e.getBody()));
        }
      }
    }
    source.stop();
  }

  /**
   * Test the reassembly of a single line across multiple packets.
   */
  @Test
  public void testFragmented() throws CharacterCodingException {
    final int maxLen = 100;

    IoBuffer savedBuf = IoBuffer.allocate(maxLen);

    String origMsg = "<1>- - blah blam foo\n";
    IoBuffer buf1 = IoBuffer.wrap(
        origMsg.substring(0, 11).getBytes(Charsets.UTF_8));
    IoBuffer buf2 = IoBuffer.wrap(
        origMsg.substring(11, 16).getBytes(Charsets.UTF_8));
    IoBuffer buf3 = IoBuffer.wrap(
        origMsg.substring(16, 21).getBytes(Charsets.UTF_8));

    LineSplitter lineSplitter = new LineSplitter(maxLen);
    ParsedBuffer parsedLine = new ParsedBuffer();

    Assert.assertFalse("Incomplete line should not be parsed",
        lineSplitter.parseLine(buf1, savedBuf, parsedLine));
    Assert.assertFalse("Incomplete line should not be parsed",
        lineSplitter.parseLine(buf2, savedBuf, parsedLine));
    Assert.assertTrue("Completed line should be parsed",
        lineSplitter.parseLine(buf3, savedBuf, parsedLine));

    // the fragmented message should now be reconstructed
    Assert.assertEquals(origMsg.trim(),
        parsedLine.buffer.getString(Charsets.UTF_8.newDecoder()));
    parsedLine.buffer.rewind();

    MultiportSyslogTCPSource.MultiportSyslogHandler handler =
        new MultiportSyslogTCPSource.MultiportSyslogHandler(maxLen, 100, null,
        null, SyslogSourceConfigurationConstants.DEFAULT_PORT_HEADER,
        new ThreadSafeDecoder(Charsets.UTF_8),
        new ConcurrentHashMap<Integer, ThreadSafeDecoder>(),
        null);

    Event event = handler.parseEvent(parsedLine, Charsets.UTF_8.newDecoder());
    String body = new String(event.getBody(), Charsets.UTF_8);
    Assert.assertEquals("Event body incorrect",
        origMsg.trim().substring(7), body);
  }

  /**
   * Test parser handling of different character sets.
   */
  @Test
  public void testCharsetParsing() throws FileNotFoundException, IOException {
    String header = "<10>2012-08-11T01:01:01Z localhost ";
    String enBody = "Yarf yarf yarf";
    String enMsg = header + enBody;
    String frBody = "Comment " + "\u00EA" + "tes-vous?";
    String frMsg = header + frBody;
    String esBody = "¿Cómo estás?";
    String esMsg = header + esBody;

    // defaults to UTF-8
    MultiportSyslogHandler handler = new MultiportSyslogHandler(
        1000, 10, new ChannelProcessor(new ReplicatingChannelSelector()),
        new SourceCounter("test"), "port",
        new ThreadSafeDecoder(Charsets.UTF_8),
        new ConcurrentHashMap<Integer, ThreadSafeDecoder>(),
        null);

    ParsedBuffer parsedBuf = new ParsedBuffer();
    parsedBuf.incomplete = false;

    // should be able to encode/decode any of these messages in UTF-8 or ISO
    String[] bodies = { enBody, esBody, frBody };
    String[] msgs = { enMsg, esMsg, frMsg };
    Charset[] charsets = { Charsets.UTF_8, Charsets.ISO_8859_1 };
    for (Charset charset : charsets) {
      for (int i = 0; i < msgs.length; i++) {
        String msg = msgs[i];
        String body = bodies[i];
        parsedBuf.buffer = IoBuffer.wrap(msg.getBytes(charset));
        Event evt = handler.parseEvent(parsedBuf, charset.newDecoder());
        String result = new String(evt.getBody(), charset);
        // this doesn't work with non-UTF-8 chars... not sure why...
        Assert.assertEquals(charset + " parse error: " + msg, body, result);
        Assert.assertNull(
            evt.getHeaders().get(SyslogUtils.EVENT_STATUS));
      }
    }

    // Construct an invalid UTF-8 sequence.
    // The parser should still generate an Event, but mark it as INVALID.
    byte[] badUtf8Seq = enMsg.getBytes(Charsets.ISO_8859_1);
    int badMsgLen = badUtf8Seq.length;
    badUtf8Seq[badMsgLen - 2] = (byte)0xFE; // valid ISO-8859-1, invalid UTF-8
    badUtf8Seq[badMsgLen - 1] = (byte)0xFF; // valid ISO-8859-1, invalid UTF-8
    parsedBuf.buffer = IoBuffer.wrap(badUtf8Seq);
    Event evt = handler.parseEvent(parsedBuf, Charsets.UTF_8.newDecoder());
    Assert.assertEquals("event body: " +
        new String(evt.getBody(), Charsets.ISO_8859_1) +
        " and my default charset = " + Charset.defaultCharset() +
        " with event = " + evt,
        SyslogUtils.SyslogStatus.INVALID.getSyslogStatus(),
        evt.getHeaders().get(SyslogUtils.EVENT_STATUS));
    Assert.assertArrayEquals("Raw message data should be kept in body of event",
        badUtf8Seq, evt.getBody());

  }

  // helper function
  private static Event takeEvent(Channel channel) {
    Transaction txn = channel.getTransaction();
    txn.begin();
    Event evt = channel.take();
    txn.commit();
    txn.close();
    return evt;
  }

  /**
   * Test that different charsets are parsed by different ports correctly.
   */
  @Test
  public void testPortCharsetHandling() throws UnknownHostException, Exception {

    ///////////////////////////////////////////////////////
    // port setup

    InetAddress localAddr = InetAddress.getLocalHost();
    DefaultIoSessionDataStructureFactory dsFactory =
        new DefaultIoSessionDataStructureFactory();


    // one faker on port 10001
    int port1 = 10001;
    NioSession session1 = mock(NioSession.class);
    session1.setAttributeMap(dsFactory.getAttributeMap(session1));
    SocketAddress sockAddr1 = new InetSocketAddress(localAddr, port1);
    when(session1.getLocalAddress()).thenReturn(sockAddr1);

    // another faker on port 10002
    int port2 = 10002;
    NioSession session2 = mock(NioSession.class);
    session2.setAttributeMap(dsFactory.getAttributeMap(session2));
    SocketAddress sockAddr2 = new InetSocketAddress(localAddr, port2);
    when(session2.getLocalAddress()).thenReturn(sockAddr2);

    // set up expected charsets per port
    ConcurrentMap<Integer, ThreadSafeDecoder> portCharsets =
        new ConcurrentHashMap<Integer, ThreadSafeDecoder>();
    portCharsets.put(port1, new ThreadSafeDecoder(Charsets.ISO_8859_1));
    portCharsets.put(port2, new ThreadSafeDecoder(Charsets.UTF_8));

    ///////////////////////////////////////////////////////
    // channel / source setup

    // set up channel to receive events
    MemoryChannel chan = new MemoryChannel();
    chan.configure(new Context());
    chan.start();
    ReplicatingChannelSelector sel = new ReplicatingChannelSelector();
    sel.setChannels(Lists.<Channel>newArrayList(chan));
    ChannelProcessor chanProc = new ChannelProcessor(sel);

    // defaults to UTF-8
    MultiportSyslogHandler handler = new MultiportSyslogHandler(
        1000, 10, chanProc, new SourceCounter("test"), "port",
        new ThreadSafeDecoder(Charsets.UTF_8), portCharsets,
        null);

    // initialize buffers
    handler.sessionCreated(session1);
    handler.sessionCreated(session2);

    ///////////////////////////////////////////////////////
    // event setup

    // Create events of varying charsets.
    String header = "<10>2012-08-17T02:14:00-07:00 192.168.1.110 ";

    // These chars encode under ISO-8859-1 as illegal bytes under UTF-8.
    String dangerousChars = "þÿÀÁ";

    ///////////////////////////////////////////////////////
    // encode and send them through the message handler
    String msg;
    IoBuffer buf;
    Event evt;

    // valid ISO-8859-1 on the right (ISO-8859-1) port
    msg = header + dangerousChars + "\n";
    buf = IoBuffer.wrap(msg.getBytes(Charsets.ISO_8859_1));
    handler.messageReceived(session1, buf);
    evt = takeEvent(chan);
    Assert.assertNotNull("Event vanished!", evt);
    Assert.assertNull(evt.getHeaders().get(SyslogUtils.EVENT_STATUS));

    // valid ISO-8859-1 on the wrong (UTF-8) port
    msg = header + dangerousChars + "\n";
    buf = IoBuffer.wrap(msg.getBytes(Charsets.ISO_8859_1));
    handler.messageReceived(session2, buf);
    evt = takeEvent(chan);
    Assert.assertNotNull("Event vanished!", evt);
    Assert.assertEquals("Expected invalid event due to character encoding",
        SyslogUtils.SyslogStatus.INVALID.getSyslogStatus(),
        evt.getHeaders().get(SyslogUtils.EVENT_STATUS));

    // valid UTF-8 on the right (UTF-8) port
    msg = header + dangerousChars + "\n";
    buf = IoBuffer.wrap(msg.getBytes(Charsets.UTF_8));
    handler.messageReceived(session2, buf);
    evt = takeEvent(chan);
    Assert.assertNotNull("Event vanished!", evt);
    Assert.assertNull(evt.getHeaders().get(SyslogUtils.EVENT_STATUS));
  }

}
