ElasticSearchJsonBolt extends BaseBasicBolt implements Configurable { void addFieldByType(XContentBuilder source, String key, String value, Map<String, String> fieldTypeMapping) throws IOException, ParseException { String fieldType = fieldTypeMapping.get(FIELD_TYPE); if (equalsIgnoreCase(fieldType, "array")) { Gson gson = new Gson(); List list = gson.fromJson(value, List.class); source.field(key, list); } else if (equalsIgnoreCase(fieldType, "date")) { String dateFormat = fieldTypeMapping.get(DATE_FORMAT); SimpleDateFormat sdf = new SimpleDateFormat(dateFormat); source.field(key, sdf.parse(value)); } else if (equalsIgnoreCase(fieldType, "microseconds")) { double microseconds = Double.parseDouble(value); double milliseconds = microseconds * 1000; source.field(key, new Date((long) milliseconds)); } else { source.field(key, value); } }  @Override void configure(Configuration conf); @Override void prepare(Map stormConf, TopologyContext context); @Override final void declareOutputFields(OutputFieldsDeclarer declarer); @Override final void execute(Tuple input, BasicOutputCollector collector); static final String ES_JSON; }
@Test public void testAddFieldByTypeArrayString(@Injectable XContentBuilder source, @Injectable String key, @Injectable Map<String, String> fieldTypeMapping) throws IOException, ParseException { new Expectations() {{ fieldTypeMapping.get(ElasticSearchJsonBolt.FIELD_TYPE); result = "array"; source.field(key, Arrays.asList("a", "b")); }}; bolt.addFieldByType(source, key, "[\"a\",\"b\"]", fieldTypeMapping); }
@Test public void testAddFieldByTypeArrayNumber(@Injectable XContentBuilder source, @Injectable String key, @Injectable Map<String, String> fieldTypeMapping) throws IOException, ParseException { new Expectations() {{ fieldTypeMapping.get(ElasticSearchJsonBolt.FIELD_TYPE); result = "array"; source.field(key, Arrays.asList(54.999444D, -1.544167D)); }}; bolt.addFieldByType(source, key, "[54.999444,-1.544167]", fieldTypeMapping); }
@Test public void testAddFieldByTypeDate(@Injectable XContentBuilder source, @Injectable String key, @Injectable Map<String, String> fieldTypeMapping) throws IOException, ParseException { Calendar instance = Calendar.getInstance(); instance.clear(); instance.setTimeZone(TimeZone.getTimeZone("UTC")); instance.set(2015, Calendar.MARCH, 14, 9, 26, 53); Date date = instance.getTime(); new Expectations() {{ fieldTypeMapping.get(ElasticSearchJsonBolt.FIELD_TYPE); result = "date"; fieldTypeMapping.get(ElasticSearchJsonBolt.DATE_FORMAT); result = "yyyy-MM-dd'T'HH:mm:ss.SSSX"; source.field(key, date); }}; bolt.addFieldByType(source, key, "2015-03-14T09:26:53.000Z", fieldTypeMapping); }
@Test public void testAddFieldByTypeMicroseconds(@Injectable XContentBuilder source, @Injectable String key, @Injectable Map<String, String> fieldTypeMapping) throws IOException, ParseException { new Expectations() {{ fieldTypeMapping.get(ElasticSearchJsonBolt.FIELD_TYPE); result = "microseconds"; source.field(key, new Date(1420082040000L)); }}; bolt.addFieldByType(source, key, "1420082040", fieldTypeMapping); }
@Test public void testAddFieldByTypeNoMapping(@Injectable XContentBuilder source, @Injectable String key, @Injectable String value) throws IOException, ParseException { new Expectations() {{ source.field(key, value); }}; bolt.addFieldByType(source, key, value, Collections.EMPTY_MAP); }