package timandes.flume.sinks;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import java.io.FileWriter;
import java.io.FileOutputStream;
import java.util.Date;
import java.text.SimpleDateFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class FileSink extends AbstractSink implements Configurable
{
    private String _pathTemplate;
    private static final Logger logger = LoggerFactory.getLogger(FileSink.class);

    @Override
    public void configure(Context context)
    {
        this._pathTemplate = context.getString("pathTemplate", "'/var/log/'YYYYMMdd'.log'");
    }

    @Override
    public void start()
    {
        logger.debug("FileSink.start()");
    }

    @Override
    public void stop()
    {
        logger.debug("FileSink.stop()");
    }

    @Override
    public Status process() throws EventDeliveryException
    {
        Status retval = null;

        Channel ch = getChannel();
        Transaction txn = ch.getTransaction();
        txn.begin();
        try {
            Event event = ch.take();
            if (event == null) {
                logger.debug("No new event was found");
                retval = Status.BACKOFF;
            } else {
                logger.debug("New events arrived");
                Date now = new Date();
                SimpleDateFormat dateFormat = new SimpleDateFormat(this._pathTemplate);
                String path = dateFormat.format(now);

                logger.debug("path=" + path);
/*
                FileWriter writer = new FileWriter(path, true);
                writer.write(event.getBody());
                writer.close();
*/
                FileOutputStream os = new FileOutputStream(path, true);
                os.write(event.getBody());
                os.write(0x0a);
                os.close();

                retval = Status.READY;
            }

            txn.commit();
        } catch (Throwable t) {
            logger.error(t.getMessage());

            txn.rollback();
            retval = Status.BACKOFF;

            if (t instanceof Error) {
                throw (Error)t;
            }
        } finally {
            txn.close();
        }

        return retval;
    }

}

@@ -137,16 +137,37 @@
 ctSink;%0A
+import java.io.File;%0A
 import j
@@ -1561,16 +1561,64 @@
  path);%0A
+%0A                this._createParentDirs(path);%0A%0A
 /*%0A     
@@ -2325,10 +2325,430 @@
 %0A    %7D%0A%0A
+    /**%0A     * Create parent dirs of given path%0A     */%0A    private void _createParentDirs(String path) throws Exception%0A    %7B%0A        File file = new File(path);%0A        String dirPath = file.getParent();%0A        %0A        File dir = new File(dirPath);%0A        if (dir.exists())%0A            return;%0A%0A        if (!dir.mkdirs())%0A            throw new Exception(%22Fail to create parent dirs for path '%22 + path + %22'%22);%0A    %7D%0A
 %7D%0A

create parent dirs when they does not exist.