@@ -0,0 +1,1636 @@
+package uk.co.samstokes.storm.serializer;%0A%0Aimport java.io.DataInputStream;%0Aimport java.io.DataOutputStream;%0Aimport java.io.InputStreamReader;%0Aimport java.io.IOException;%0Aimport java.io.OutputStreamWriter;%0Aimport java.io.Reader;%0Aimport java.io.Writer;%0A%0Aimport org.json.simple.JSONArray;%0Aimport org.json.simple.JSONObject;%0Aimport org.json.simple.JSONValue;%0Aimport org.json.simple.parser.ParseException;%0A%0Aimport backtype.storm.serialization.ISerialization;%0A%0Apublic class JSONSerializer implements ISerialization%3CObject%3E %7B%0A    public static final String ENCODING = %22UTF-8%22;%0A%0A    @SuppressWarnings(%22rawtypes%22)%0A    @Override%0A    public boolean accept(Class c) %7B%0A        return JSONObject.class.equals(c) %7C%7C%0A          JSONArray.class.equals(c);%0A    %7D%0A%0A    @Override%0A    public void serialize(Object object, DataOutputStream stream)%0A            throws IOException %7B%0A        final Writer writer = new OutputStreamWriter(stream, ENCODING);%0A        if (object instanceof JSONObject) %7B%0A            ((JSONObject) object).writeJSONString(writer);%0A        %7D else if (object instanceof JSONArray) %7B%0A            ((JSONArray) object).writeJSONString(writer);%0A        %7D else %7B%0A            throw new IllegalArgumentException(%22Unexpected class %22 + object.getClass().getCanonicalName());%0A        %7D%0A        writer.flush();%0A    %7D%0A%0A    @Override%0A    public Object deserialize(DataInputStream stream) throws IOException %7B%0A        final Reader reader = new InputStreamReader(stream, ENCODING);%0A        try %7B%0A            return JSONValue.parseWithException(reader);%0A        %7D catch (ParseException e) %7B%0A            throw new RuntimeException(e);%0A        %7D%0A    %7D%0A%7D%0A

Add a serialiser so values from JSONScheme can make it through