/*
 * 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.camel.component.hdfs;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.InputStream;
import java.net.URL;

import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.util.IOHelper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ArrayFile;
import org.apache.hadoop.io.BloomMapFile;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.ByteWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapFile;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import static org.apache.camel.language.simple.SimpleLanguage.simple;

public class HdfsProducerTest extends HdfsTestSupport {

    private static final Path TEMP_DIR = new Path(new File("target/test/").getAbsolutePath());

    @Override
    @Before
    public void setUp() throws Exception {
        if (skipTest()) {
            return;
        }
        super.setUp();
    }

    @Test
    public void testProducer() throws Exception {
        if (skipTest()) {
            return;
        }
        template.sendBody("direct:start1", "PAPPO");

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel1");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        assertEquals("PAPPO", value.toString());

        IOHelper.close(reader);
    }

    @Test
    public void testProducerClose() throws Exception {
        if (skipTest()) {
            return;
        }
        for (int i = 0; i < 10; ++i) {
            // send 10 messages, and mark to close in last message
            template.sendBodyAndHeader("direct:start1", "PAPPO" + i, HdfsConstants.HDFS_CLOSE, i == 9 ? true : false);
        }

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel1");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);

        int i = 0;
        while (reader.next(key, value)) {
            Text txt = (Text) value;
            assertEquals("PAPPO" + i, txt.toString());
            ++i;
        }

        IOHelper.close(reader);
    }

    @Test
    public void testWriteBoolean() throws Exception {
        if (skipTest()) {
            return;
        }
        Boolean aBoolean = true;
        template.sendBody("direct:write_boolean", aBoolean);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-boolean");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        Boolean rBoolean = ((BooleanWritable) value).get();
        assertEquals(rBoolean, aBoolean);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteByte() throws Exception {
        if (skipTest()) {
            return;
        }
        byte aByte = 8;
        template.sendBody("direct:write_byte", aByte);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-byte");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        byte rByte = ((ByteWritable) value).get();
        assertEquals(rByte, aByte);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteInt() throws Exception {
        if (skipTest()) {
            return;
        }
        int anInt = 1234;
        template.sendBody("direct:write_int", anInt);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-int");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        int rInt = ((IntWritable) value).get();
        assertEquals(rInt, anInt);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteFloat() throws Exception {
        if (skipTest()) {
            return;
        }
        float aFloat = 12.34f;
        template.sendBody("direct:write_float", aFloat);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-float");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        float rFloat = ((FloatWritable) value).get();
        assertEquals(rFloat, aFloat, 0.0F);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteDouble() throws Exception {
        if (skipTest()) {
            return;
        }
        Double aDouble = 12.34D;
        template.sendBody("direct:write_double", aDouble);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-double");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        Double rDouble = ((DoubleWritable) value).get();
        assertEquals(rDouble, aDouble);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteLong() throws Exception {
        if (skipTest()) {
            return;
        }
        long aLong = 1234567890;
        template.sendBody("direct:write_long", aLong);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-long");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        long rLong = ((LongWritable) value).get();
        assertEquals(rLong, aLong);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteText() throws Exception {
        if (skipTest()) {
            return;
        }
        String txt = "CIAO MONDO !";
        template.sendBody("direct:write_text1", txt);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-text1");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Text value = (Text) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        String rTxt = value.toString();
        assertEquals(rTxt, txt);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteTextWithKey() throws Exception {
        if (skipTest()) {
            return;
        }
        String txtKey = "THEKEY";
        String txtValue = "CIAO MONDO !";
        template.sendBodyAndHeader("direct:write_text2", txtValue, "KEY", txtKey);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-text2");
        SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file1));
        Text key = (Text) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Text value = (Text) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        assertEquals(key.toString(), txtKey);
        assertEquals(value.toString(), txtValue);

        IOHelper.close(reader);
    }

    @Test
    public void testMapWriteTextWithKey() throws Exception {
        if (skipTest()) {
            return;
        }
        String txtKey = "THEKEY";
        String txtValue = "CIAO MONDO !";
        template.sendBodyAndHeader("direct:write_text3", txtValue, "KEY", txtKey);

        Configuration conf = new Configuration();
        MapFile.Reader reader = new MapFile.Reader(new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-text3"), conf);
        Text key = (Text) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Text value = (Text) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        assertEquals(key.toString(), txtKey);
        assertEquals(value.toString(), txtValue);

        IOHelper.close(reader);
    }

    @Test
    public void testArrayWriteText() throws Exception {
        if (skipTest()) {
            return;
        }
        String txtValue = "CIAO MONDO !";
        template.sendBody("direct:write_text4", txtValue);

        Configuration conf = new Configuration();
        Path file1 = new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-text4");
        FileSystem fs1 = FileSystem.get(file1.toUri(), conf);
        ArrayFile.Reader reader = new ArrayFile.Reader(fs1, "file:///" + TEMP_DIR.toUri() + "/test-camel-text4", conf);
        Text value = (Text) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(value);
        assertEquals(value.toString(), txtValue);

        IOHelper.close(reader);
    }

    @Test
    public void testBloomMapWriteText() throws Exception {
        if (skipTest()) {
            return;
        }
        String txtKey = "THEKEY";
        String txtValue = "CIAO MONDO !";
        template.sendBodyAndHeader("direct:write_text5", txtValue, "KEY", txtKey);

        Configuration conf = new Configuration();
        BloomMapFile.Reader reader = new BloomMapFile.Reader(new Path("file:///" + TEMP_DIR.toUri() + "/test-camel-text5"), conf);
        Text key = (Text) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
        Text value = (Text) ReflectionUtils.newInstance(reader.getValueClass(), conf);
        reader.next(key, value);
        assertEquals(key.toString(), txtKey);
        assertEquals(value.toString(), txtValue);

        IOHelper.close(reader);
    }

    @Test
    public void testWriteTextWithDynamicFilename() throws Exception {
        if (skipTest()) {
            return;
        }

        for (int i = 0; i < 5; i++) {
            template.sendBodyAndHeader("direct:write_dynamic_filename", "CIAO" + i, Exchange.FILE_NAME, "file" + i);
        }

        for (int i = 0; i < 5; i++) {
            InputStream in = null;
            try {
                in = new URL("file:///" + TEMP_DIR.toUri() + "/test-camel-dynamic/file" + i).openStream();
                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                IOUtils.copyBytes(in, bos, 4096, false);
                assertEquals("CIAO" + i, new String(bos.toByteArray()));
            } finally {
                IOHelper.close(in);
            }
        }
    }

    @Test
    public void testWriteTextWithDynamicFilenameExpression() throws Exception {
        if (skipTest()) {
            return;
        }

        for (int i = 0; i < 5; i++) {
            template.sendBodyAndHeader("direct:write_dynamic_filename", "CIAO" + i, Exchange.FILE_NAME, simple("file-${body}"));
        }

        for (int i = 0; i < 5; i++) {
            InputStream in = null;
            try {
                in = new URL("file:///" + TEMP_DIR.toUri() + "/test-camel-dynamic/file-CIAO" + i).openStream();
                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                IOUtils.copyBytes(in, bos, 4096, false);
                assertEquals("CIAO" + i, new String(bos.toByteArray()));
            } finally {
                IOHelper.close(in);
            }
        }
    }

    @Override
    @After
    public void tearDown() throws Exception {
        if (skipTest()) {
            return;
        }
        super.tearDown();

        Thread.sleep(250);
        Configuration conf = new Configuration();
        Path dir = new Path("target/test");
        FileSystem fs = FileSystem.get(dir.toUri(), conf);
        fs.delete(dir, true);
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {

            @Override
            public void configure() throws Exception {
                from("direct:start1").to("hdfs://localhost/" + TEMP_DIR.toUri() + "/test-camel1?fileSystemType=LOCAL&valueType=TEXT&fileType=SEQUENCE_FILE");

                /* For testing writables */
                from("direct:write_boolean").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-boolean?fileSystemType=LOCAL&valueType=BOOLEAN&fileType=SEQUENCE_FILE");

                from("direct:write_byte").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-byte?fileSystemType=LOCAL&valueType=BYTE&fileType=SEQUENCE_FILE");

                from("direct:write_int").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-int?fileSystemType=LOCAL&valueType=INT&fileType=SEQUENCE_FILE");

                from("direct:write_float").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-float?fileSystemType=LOCAL&valueType=FLOAT&fileType=SEQUENCE_FILE");

                from("direct:write_long").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-long?fileSystemType=LOCAL&valueType=LONG&fileType=SEQUENCE_FILE");

                from("direct:write_double").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-double?fileSystemType=LOCAL&valueType=DOUBLE&fileType=SEQUENCE_FILE");

                from("direct:write_text1").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-text1?fileSystemType=LOCAL&valueType=TEXT&fileType=SEQUENCE_FILE");

                /* For testing key and value writing */
                from("direct:write_text2").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-text2?fileSystemType=LOCAL&keyType=TEXT&valueType=TEXT&fileType=SEQUENCE_FILE");

                from("direct:write_text3").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-text3?fileSystemType=LOCAL&keyType=TEXT&valueType=TEXT&fileType=MAP_FILE");

                /* For testing ArrayFile */
                from("direct:write_text4").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-text4?fileSystemType=LOCAL&valueType=TEXT&fileType=ARRAY_FILE");

                /* For testing BloomMapFile */
                from("direct:write_text5").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-text5?fileSystemType=LOCAL&keyType=TEXT&valueType=TEXT&fileType=BLOOMMAP_FILE");

                from("direct:write_dynamic_filename").to("hdfs:localhost/" + TEMP_DIR.toUri() + "/test-camel-dynamic/?fileSystemType=LOCAL&valueType=TEXT");
            }
        };
    }
}