JSONHandler implements HTTPSourceHandler { @Override public List<Event> getEvents(HttpServletRequest request) throws Exception { BufferedReader reader = request.getReader(); String charset = request.getCharacterEncoding(); if (charset == null) { LOG.debug("Charset is null, default charset of UTF-8 will be used."); charset = "UTF-8"; } else if (!(charset.equalsIgnoreCase("utf-8") || charset.equalsIgnoreCase("utf-16") || charset.equalsIgnoreCase("utf-32"))) { LOG.error("Unsupported character set in request {}. " + "JSON handler supports UTF-8, " + "UTF-16 and UTF-32 only.", charset); throw new UnsupportedCharsetException("JSON handler supports UTF-8, " + "UTF-16 and UTF-32 only."); } List<Event> eventList = new ArrayList<Event>(0); try { eventList = gson.fromJson(reader, listType); } catch (JsonSyntaxException ex) { throw new HTTPBadRequestException("Request has invalid JSON Syntax.", ex); } for (Event e : eventList) { ((JSONEvent) e).setCharset(charset); } return getSimpleEvents(eventList); } JSONHandler(); @Override List<Event> getEvents(HttpServletRequest request); @Override void configure(Context context);  }
@Test public void testMultipleEvents() throws Exception { String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(),"UTF-8")); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("random_body2", new String(e.getBody(),"UTF-8")); }
@Test public void testMultipleEventsUTF16() throws Exception { String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-16"); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(), "UTF-16")); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-16")); }
@Test public void testMultipleEventsUTF32() throws Exception { String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-32"); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(), "UTF-32")); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-32")); }
@Test public void testMultipleEventsUTF8() throws Exception { String json = "[{\"headers\":{\"a\": \"b\"},\"body\": \"random_body\"}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"random_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "UTF-8"); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(), "UTF-8")); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("random_body2", new String(e.getBody(), "UTF-8")); }
@Test public void testEscapedJSON() throws Exception { String json = "[{\"headers\":{\"a\": \"b\"}}," + "{\"headers\":{\"e\": \"f\"},\"body\": \"rand\\\"om_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertTrue(e.getBody().length == 0); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("rand\"om_body2", new String(e.getBody(),"UTF-8")); }
@Test public void testNoBody() throws Exception { String json = "[{\"headers\" : {\"a\": \"b\"}}," + "{\"headers\" : {\"e\": \"f\"},\"body\": \"random_body2\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertTrue(e.getBody().length == 0); e = deserialized.get(1); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("random_body2", new String(e.getBody(),"UTF-8")); }
@Test public void testSingleHTMLEvent() throws Exception { String json = "[{\"headers\": {\"a\": \"b\"}," + "\"body\": \"<html><body>test</body></html>\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("<html><body>test</body></html>", new String(e.getBody(),"UTF-8")); }
@Test public void testSingleEvent() throws Exception { String json = "[{\"headers\" : {\"a\": \"b\"},\"body\": \"random_body\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(),"UTF-8")); }
@Test(expected = HTTPBadRequestException.class) public void testBadEvent() throws Exception { String json = "{[\"a\": \"b\"],\"body\": \"random_body\"}"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); handler.getEvents(req); Assert.fail(); }
@Test(expected = UnsupportedCharsetException.class) public void testError() throws Exception { String json = "[{\"headers\" : {\"a\": \"b\"},\"body\": \"random_body\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json, "ISO-8859-1"); handler.getEvents(req); Assert.fail(); }
@Test public void testSingleEventInArray() throws Exception { String json = "[{\"headers\": {\"a\": \"b\"},\"body\": \"random_body\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("random_body", new String(e.getBody(),"UTF-8")); }
@Test public void testMultipleLargeEvents() throws Exception { String json = "[{\"headers\" : {\"a\": \"b\", \"a2\": \"b2\"," + "\"a3\": \"b3\",\"a4\": \"b4\"},\"body\": \"random_body\"}," + "{\"headers\" :{\"e\": \"f\",\"e2\": \"f2\"," + "\"e3\": \"f3\",\"e4\": \"f4\",\"e5\": \"f5\"}," + "\"body\": \"random_body2\"}," + "{\"headers\" :{\"q1\": \"b\",\"q2\": \"b2\",\"q3\": \"b3\",\"q4\": \"b4\"}," + "\"body\": \"random_bodyq\"}]"; HttpServletRequest req = new FlumeHttpServletRequestWrapper(json); List<Event> deserialized = handler.getEvents(req); Event e = deserialized.get(0); Assert.assertNotNull(e); Assert.assertEquals("b", e.getHeaders().get("a")); Assert.assertEquals("b2", e.getHeaders().get("a2")); Assert.assertEquals("b3", e.getHeaders().get("a3")); Assert.assertEquals("b4", e.getHeaders().get("a4")); Assert.assertEquals("random_body", new String(e.getBody(),"UTF-8")); e = deserialized.get(1); Assert.assertNotNull(e); Assert.assertEquals("f", e.getHeaders().get("e")); Assert.assertEquals("f2", e.getHeaders().get("e2")); Assert.assertEquals("f3", e.getHeaders().get("e3")); Assert.assertEquals("f4", e.getHeaders().get("e4")); Assert.assertEquals("f5", e.getHeaders().get("e5")); Assert.assertEquals("random_body2", new String(e.getBody(),"UTF-8")); e = deserialized.get(2); Assert.assertNotNull(e); Assert.assertEquals("b", e.getHeaders().get("q1")); Assert.assertEquals("b2", e.getHeaders().get("q2")); Assert.assertEquals("b3", e.getHeaders().get("q3")); Assert.assertEquals("b4", e.getHeaders().get("q4")); Assert.assertEquals("random_bodyq", new String(e.getBody(),"UTF-8")); }
@Test public void testDeserializarion() throws Exception { Type listType = new TypeToken<List<JSONEvent>>() { }.getType(); List<JSONEvent> events = Lists.newArrayList(); Random rand = new Random(); for (int i = 1; i < 10; i++) { Map<String, String> input = Maps.newHashMap(); for (int j = 1; j < 10; j++) { input.put(String.valueOf(i) + String.valueOf(j), String.valueOf(i)); } JSONEvent e = new JSONEvent(); e.setBody(String.valueOf(rand.nextGaussian()).getBytes("UTF-8")); e.setHeaders(input); events.add(e); } Gson gson = new Gson(); List<Event> deserialized = handler.getEvents( new FlumeHttpServletRequestWrapper(gson.toJson(events, listType))); int i = 0; for (Event e : deserialized) { Event current = events.get(i++); Assert.assertEquals(new String(current.getBody(),"UTF-8"), new String(e.getBody(),"UTF-8")); Assert.assertEquals(current.getHeaders(), e.getHeaders()); } }