/*
 * Copyright 2011 Ning, Inc.
 *
 * Ning 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 com.ning.metrics.serialization.event;

import com.ning.metrics.serialization.smile.SmileBucket;
import com.ning.metrics.serialization.smile.SmileBucketDeserializer;
import com.ning.metrics.serialization.smile.SmileBucketSerializer;
import org.joda.time.DateTime;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.nio.charset.Charset;

/**
 * A meta-event wrapping a collector of Smile events of the same type and same granularity.
 * This is used for serializing, to leverage compression (back-references) that Smile offers.
 */
public class SmileBucketEvent implements Event
{
    private String eventName;
    private Granularity granularity;
    private String suffixOutputPath = null;
    private SmileBucket bucket;

    private ByteArrayOutputStream eventStream = null;
    private static final Charset CHARSET = Charset.forName("UTF-8");

    public SmileBucketEvent(String eventName, Granularity granularity, SmileBucket bucket)
    {
        this(eventName, granularity, null, bucket);
    }

    @Deprecated
    public SmileBucketEvent()
    {
    }

    public SmileBucketEvent(String eventName, Granularity granularity, String baseOutputDir, SmileBucket bucket)
    {
        this.eventName = eventName;
        this.granularity = granularity;
        this.suffixOutputPath = baseOutputDir;
        this.bucket = bucket;
    }

    public SmileBucket getBucket()
    {
        return bucket;
    }

    @Override
    public DateTime getEventDateTime()
    {
        throw new RuntimeException("Events in SmileBucket have different timestamps!");
    }

    @Override
    public String getName()
    {
        return eventName;
    }

    @Override
    public Granularity getGranularity()
    {
        return granularity;
    }

    @Override
    public String getVersion()
    {
        return null;
    }

    @Override
    public String getOutputDir(String prefix)
    {
        if (suffixOutputPath == null) {
            // Add a safeguard here - if it's not set, the caller must be doing something wrong
            throw new RuntimeException("suffixOutputPath not set, events not properly grouped. See SmileEnvelopeEventsToSmileBucketEvents class.");
        }
        else {
            return String.format("%s%s", prefix, suffixOutputPath);
        }
    }

    /**
     * @return Object representing the data, compressed Smile version of the events
     */
    @Override
    public Object getData()
    {
        if (eventStream == null) {
            // Start with 16kB buffer
            eventStream = new ByteArrayOutputStream(16384);

            try {
                SmileBucketSerializer.serialize(bucket, eventStream);
            }
            catch (IOException e) {
                return null;
            }
        }

        return eventStream;
    }

    /**
     * Serialize an event to a byte array.
     * This method is optional, methods relying on this call should handle gracefully null.
     *
     * @return byte array representation of an event, can return null
     */
    @Override
    public byte[] getSerializedEvent()
    {
        return ((ByteArrayOutputStream) getData()).toByteArray();
    }

    @Override
    public void writeExternal(ObjectOutput objectOutput) throws IOException
    {
        int eventNameLength = eventName.length();
        objectOutput.writeInt(eventNameLength);
        objectOutput.write(eventName.getBytes(CHARSET));

        int granularityLength = granularity.toString().length();
        objectOutput.writeInt(granularityLength);
        objectOutput.write(granularity.toString().getBytes(CHARSET));

        byte[] data = getSerializedEvent();
        int dataLen = data.length;

        objectOutput.writeInt(dataLen);
        objectOutput.write(data);
    }

    @Override
    public void readExternal(ObjectInput objectInput) throws IOException, ClassNotFoundException
    {
        int eventNameLength = objectInput.readInt();
        byte[] eventNameData = new byte[eventNameLength];
        objectInput.readFully(eventNameData);
        eventName = new String(eventNameData, CHARSET);

        int granularityLength = objectInput.readInt();
        byte[] granularityData = new byte[granularityLength];
        objectInput.readFully(granularityData);
        try {
            granularity = Granularity.valueOf(new String(granularityData, CHARSET));
        }
        catch (IllegalArgumentException e) {
            granularity = Granularity.HOURLY;
        }

        int dataLen = objectInput.readInt();
        byte[] data = new byte[dataLen];
        objectInput.readFully(data);

        bucket = SmileBucketDeserializer.deserialize(new ByteArrayInputStream(data));
    }

    public int getNumberOfEvent()
    {
        return bucket.size();
    }
}

@@ -1426,21 +1426,96 @@
 tPath = 
-null;
+%22%22; // Not necessarily set, and avoid triggering an NPE in serialization methods
 %0A    pri
@@ -1798,20 +1798,18 @@
 larity, 
-null
+%22%22
 , bucket
@@ -2730,24 +2730,30 @@
 tputPath
- == null
+.length() == 0
 ) %7B%0A    
@@ -4401,32 +4401,155 @@
 tes(CHARSET));%0A%0A
+        objectOutput.writeInt(suffixOutputPath.length());%0A        objectOutput.write(suffixOutputPath.getBytes(CHARSET));%0A%0A
         byte%5B%5D d
@@ -5389,16 +5389,16 @@
 HOURLY;%0A
-
         
@@ -5392,32 +5392,285 @@
 RLY;%0A        %7D%0A%0A
+        int suffixOutputPathLength = objectInput.readInt();%0A        byte%5B%5D suffixOuputPathData = new byte%5BsuffixOutputPathLength%5D;%0A        objectInput.readFully(suffixOuputPathData);%0A        suffixOutputPath = new String(suffixOuputPathData, CHARSET);%0A%0A
         int data

make sure to serialize output path of SmileBucketEvent