RegexExtractorInterceptor implements Interceptor { @Override public Event intercept(Event event) { Matcher matcher = regex.matcher( new String(event.getBody(), Charsets.UTF_8)); Map<String, String> headers = event.getHeaders(); if (matcher.find()) { for (int group = 0, count = matcher.groupCount(); group < count; group++) { int groupIndex = group + 1; if (groupIndex > serializers.size()) { if (logger.isDebugEnabled()) { logger.debug("Skipping group {} to {} due to missing serializer", group, count); } break; } NameAndSerializer serializer = serializers.get(group); if (logger.isDebugEnabled()) { logger.debug("Serializing {} using {}", serializer.headerName, serializer.serializer); } headers.put(serializer.headerName, serializer.serializer.serialize(matcher.group(groupIndex))); } } return event; } private  RegexExtractorInterceptor(Pattern regex,
      List<NameAndSerializer> serializers); @Override void initialize(); @Override void close(); @Override Event intercept(Event event); @Override List<Event> intercept(List<Event> events);  }
@Test public void shouldExtractAddHeadersForAllMatchGroups() throws Exception { Context context = new Context(); context.put(RegexExtractorInterceptor.REGEX, "(\\d):(\\d):(\\d)"); context.put(RegexExtractorInterceptor.SERIALIZERS, "s1 s2 s3"); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s1.name", "Num1"); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s2.name", "Num2"); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s3.name", "Num3"); fixtureBuilder.configure(context); Interceptor fixture = fixtureBuilder.build(); Event event = EventBuilder.withBody("1:2:3.4foobar5", Charsets.UTF_8); Event expected = EventBuilder.withBody("1:2:3.4foobar5", Charsets.UTF_8); expected.getHeaders().put("Num1", "1"); expected.getHeaders().put("Num2", "2"); expected.getHeaders().put("Num3", "3"); Event actual = fixture.intercept(event); Assert.assertArrayEquals(expected.getBody(), actual.getBody()); Assert.assertEquals(expected.getHeaders(), actual.getHeaders()); }
@Test public void shouldExtractAddHeadersForAllMatchGroupsIgnoringMissingIds() throws Exception { String body = "2012-10-17 14:34:44,338"; Context context = new Context(); context.put(RegexExtractorInterceptor.REGEX, "^(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)(:\\d\\d,\\d\\d\\d)"); context.put(RegexExtractorInterceptor.SERIALIZERS, "s1"); context .put(RegexExtractorInterceptor.SERIALIZERS + ".s1.name", "timestamp"); fixtureBuilder.configure(context); Interceptor fixture = fixtureBuilder.build(); Event event = EventBuilder.withBody(body, Charsets.UTF_8); Event expected = EventBuilder.withBody(body, Charsets.UTF_8); expected.getHeaders().put("timestamp", "2012-10-17 14:34"); Event actual = fixture.intercept(event); Assert.assertArrayEquals(expected.getBody(), actual.getBody()); Assert.assertEquals(expected.getHeaders(), actual.getHeaders()); }
@Test public void shouldExtractAddHeadersUsingSpecifiedSerializer() throws Exception { long now = (System.currentTimeMillis() / 60000L) * 60000L; String pattern = "yyyy-MM-dd HH:mm:ss,SSS"; DateTimeFormatter formatter = DateTimeFormat.forPattern(pattern); String body = formatter.print(now); System.out.println(body); Context context = new Context(); context.put(RegexExtractorInterceptor.REGEX, "^(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)(:\\d\\d,\\d\\d\\d)"); context.put(RegexExtractorInterceptor.SERIALIZERS, "s1 s2"); String millisSerializers = RegexExtractorInterceptorMillisSerializer.class.getName(); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s1.type", millisSerializers); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s1.name", "timestamp"); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s1.pattern", "yyyy-MM-dd HH:mm"); context.put(RegexExtractorInterceptor.SERIALIZERS + ".s2.name", "data"); fixtureBuilder.configure(context); Interceptor fixture = fixtureBuilder.build(); Event event = EventBuilder.withBody(body, Charsets.UTF_8); Event expected = EventBuilder.withBody(body, Charsets.UTF_8); expected.getHeaders().put("timestamp", String.valueOf(now)); expected.getHeaders().put("data", ":00,000"); Event actual = fixture.intercept(event); Assert.assertArrayEquals(expected.getBody(), actual.getBody()); Assert.assertEquals(expected.getHeaders(), actual.getHeaders()); }