/**
 * Amazon Kinesis Aggregators
 *
 * Copyright 2014, Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Amazon Software License (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/asl/
 *
 * or in the "license" file accompanying this file. This file 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.amazonaws.services.kinesis.io.serializer;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;

import com.amazonaws.services.kinesis.aggregators.InputEvent;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;

@SuppressWarnings("rawtypes")
/**
 * Class which handles serialising Object payloads using Jackson marshalling, or converts to string format if configured to support text based payloads
 */
public class JsonSerializer implements IKinesisSerializer<Object, byte[]> {
    ObjectMapper mapper = new ObjectMapper();

    String itemTerminator = null;

    Class clazz;

    private String filterRegex;

    private Pattern p;

    private String charset = "UTF-8";

    /**
     * Construct a basic json data serialiser
     */
    public JsonSerializer() {
    }

    /**
     * Construct a serialiser that is based on a densely packed recordset list
     * of items
     * 
     * @param itemTerminator
     */
    public JsonSerializer(String itemTerminator) {
        this.itemTerminator = itemTerminator;
    }

    /**
     * Construct a Serialiser which is class based
     * 
     * @param clazz
     */
    public JsonSerializer(Class clazz) {
        this.clazz = clazz;
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    @SuppressWarnings("unchecked")
    /**
     * Method to generate either a class instance from a Kinesis Record, or a String which will be converted to JsonMap if we are serialising text based payloads
     */
    public Object toClass(final InputEvent event) throws IOException {
        // Return a class object from the json, or if we have no class then
        // return a String list
        List<String> jsonStringList = new ArrayList<>();

        if (this.clazz == null) {
            if (this.itemTerminator != null) {
                // break up the json items as separate lines
                String[] items = new String(event.getData(), this.charset).split(this.itemTerminator);

                for (String item : items) {
                    if (filterRegex == null || (filterRegex != null && p.matcher(item).matches())) {
                        jsonStringList.add(item);
                    }
                }

                return jsonStringList;
            } else {
                // single json object per record
                String item = new String(event.getData(), this.charset);

                if (filterRegex == null || (filterRegex != null && p.matcher(item).matches())) {
                    jsonStringList.add(item);
                }

                return jsonStringList;
            }
        } else {
            // use jackson to serialise a class instance
            return mapper.readValue(event.getData(), clazz);
        }
    }

    /**
     * Convert a given object into the required binary representation, based
     * upon the serialiser config as either an object serialiser or a string
     * serialiser
     */
    public byte[] fromClass(final Object o) throws IOException {
        if (this.clazz == null) {
            return SerializationUtils.safeReturnData(((String) o).getBytes(this.charset));
        } else {
            return SerializationUtils.safeReturnData(mapper.writeValueAsBytes(o));
        }
    }

    /**
     * Builder method to apply a filtering regular expression to text based
     * serialisation operations
     * 
     * @param regex
     * @return
     */
    public JsonSerializer withFilterRegex(String regex) {
        this.filterRegex = regex;
        p = Pattern.compile(this.filterRegex);

        return this;
    }

    /**
     * Builder method to apply a non-default character set to text based
     * serialisation operations (default UTF-8)
     * 
     * @param charset
     * @return
     */
    public JsonSerializer withCharset(String charset) {
        // test that this is a valid character set
        Charset test = Charset.forName(charset);

        // use it
        this.charset = charset;

        return this;
    }

    /**
     * Build method to apply a non-default item terminator (default \n)
     * 
     * @param itemTerminator
     * @return
     */
    public JsonSerializer withItemTerminator(String itemTerminator) {
        this.itemTerminator = itemTerminator;
        return this;
    }
}

@@ -739,24 +739,51 @@
 .ArrayList;%0A
+import java.util.Iterator;%0A
 import java.
@@ -789,24 +789,24 @@
 .util.List;%0A
-
 import java.
@@ -888,16 +888,173 @@
 tEvent;%0A
+import com.amazonaws.services.kinesis.aggregators.StreamAggregatorUtils;%0Aimport com.amazonaws.services.kinesis.aggregators.exception.SerializationException;%0A
 import c
@@ -1107,16 +1107,64 @@
 eature;%0A
+import com.fasterxml.jackson.databind.JsonNode;%0A
 import c
@@ -3383,32 +3383,398 @@
 this.charset);%0A%0A
+                if (item.charAt(0) == '%5B') %7B%0A                    try %7B%0A                        JsonNode json = StreamAggregatorUtils.asJsonNode(item);%0A                        for (Iterator%3CJsonNode%3E iter = json.elements(); iter.hasNext();) %7B%0A                            JsonNode child = iter.next();%0A                            item = child.toString();%0A%0A            
                 
@@ -3866,32 +3866,44 @@
                 
+            
 jsonStringList.a
@@ -3904,32 +3904,418 @@
 List.add(item);%0A
+                            %7D%0A                        %7D%0A                    %7D catch (Exception e) %7B%0A                        throw new IOException(e);%0A                    %7D%0A                %7D%0A                else %7B%0A                    if (filterRegex == null %7C%7C (filterRegex != null && p.matcher(item).matches())) %7B%0A                        jsonStringList.add(item);%0A                    %7D%0A
                 

Handle JSON arrays