diff --git a/pom.xml b/pom.xml
index 6b09d09..2b501fb 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,6 +20,11 @@
amqp-client
2.6.1
+
+ com.googlecode.json-simple
+ json-simple
+ 1.1
+
diff --git a/src/main/java/uk/co/samstokes/storm/TestTopology.java b/src/main/java/uk/co/samstokes/storm/TestTopology.java
index 9eb959f..511833e 100644
--- a/src/main/java/uk/co/samstokes/storm/TestTopology.java
+++ b/src/main/java/uk/co/samstokes/storm/TestTopology.java
@@ -6,6 +6,7 @@
import backtype.storm.topology.TopologyBuilder;
+import uk.co.samstokes.storm.scheme.JSONScheme;
import uk.co.samstokes.storm.spout.AMQPSpout;
public class TestTopology {
@@ -21,7 +22,8 @@ public static void main (String[] args) {
args[3],
args[4],
args[5],
- args[6]
+ args[6],
+ new JSONScheme()
));
final Config config = new Config();
diff --git a/src/main/java/uk/co/samstokes/storm/scheme/JSONScheme.java b/src/main/java/uk/co/samstokes/storm/scheme/JSONScheme.java
new file mode 100644
index 0000000..6049a6c
--- /dev/null
+++ b/src/main/java/uk/co/samstokes/storm/scheme/JSONScheme.java
@@ -0,0 +1,46 @@
+package uk.co.samstokes.storm.scheme;
+
+import java.io.UnsupportedEncodingException;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.json.simple.JSONValue;
+
+import backtype.storm.spout.Scheme;
+
+import backtype.storm.tuple.Fields;
+
+
+public class JSONScheme implements Scheme {
+ private static final long serialVersionUID = -7734176307841199017L;
+
+ private final String encoding;
+
+
+ public JSONScheme(String encoding) {
+ this.encoding = encoding;
+ }
+ public JSONScheme() {
+ this("UTF-8");
+ }
+
+
+ @Override
+ public List