ColumnFilteringFormattingInterceptor implements Interceptor { @Override public Event intercept(Event event) { byte[] body = event.getBody(); if (body == null) { return event; } final int length = body.length; int i = 0; int index = 0; while (i < length) { if (body[i++] == srcSeparator) { positions[++index] = i; if (index >= maxIndex) { break; } } } int nextVirginIndex; boolean separatorAtEnd = true; if (i == length && index < maxColumn) { nextVirginIndex = index + 2; positions[nextVirginIndex - 1] = length; separatorAtEnd = length > 0 ? body[length - 1] == srcSeparator : false; } else { nextVirginIndex = index + 1; } int newArrayLen = prefix.length; for (i = columns.length; i-- > 0; ) { int column = columns[i]; int len = positions[column + 1] - positions[column]; if (len > 0) { if (positions[column + 1] == length && !separatorAtEnd) { newArrayLen += len; } else { newArrayLen += len - 1; } } newArrayLen += dstSeparators[i].length; } byte[] newBody = new byte[newArrayLen]; int newOffset = 0; if (prefix.length > 0) { System.arraycopy(prefix, 0, newBody, 0, prefix.length); newOffset += prefix.length; } int dstSeparatorsIdx = 0; for (int column : columns) { int len = positions[column + 1] - positions[column]; byte[] separator = dstSeparators[dstSeparatorsIdx++]; if (len > 0) { System.arraycopy(body, positions[column], newBody, newOffset, len); newOffset += len; if (newBody[newOffset - 1] == srcSeparator) { newOffset--; } } System.arraycopy(separator, 0, newBody, newOffset, separator.length); newOffset += separator.length; } event.setBody(newBody); Arrays.fill(positions, 1, nextVirginIndex, 0); return event; } private  ColumnFilteringFormattingInterceptor(int[] columns, byte srcSeparator, byte[][] dstSeparators, byte[] prefix); @Override void initialize(); @Override Event intercept(Event event); @Override List<Event> intercept(List<Event> events); @Override void close();  }
@Test public void testInterceptEventWithPrefix() { HashMap<String, String> contextMap = new HashMap<String, String>(); contextMap.put(ColumnFilteringInterceptor.Constants.SRC_SEPARATOR, Byte.toString((byte)2)); contextMap.put(ColumnFilteringFormattingInterceptor.Constants.COLUMNS_FORMATTER, "\001{1}\001{2}\001{3}\001"); ColumnFilteringFormattingInterceptor.Builder builder = new ColumnFilteringFormattingInterceptor.Builder(); builder.configure(new Context(contextMap)); Interceptor interceptor = builder.build(); assertArrayEquals("Six Fields", "\001\001Second\001\001".getBytes(), interceptor.intercept( new InterceptorTestHelper.MyEvent("First\002\002Second\002\002\002".getBytes())).getBody()); }
@Test public void testInterceptEventWithLongSeparator() { HashMap<String, String> contextMap = new HashMap<String, String>(); contextMap.put(ColumnFilteringInterceptor.Constants.SRC_SEPARATOR, Byte.toString((byte)2)); contextMap.put(ColumnFilteringFormattingInterceptor.Constants.COLUMNS_FORMATTER, "a{1}bc{2}def{3}ghi"); ColumnFilteringFormattingInterceptor.Builder builder = new ColumnFilteringFormattingInterceptor.Builder(); builder.configure(new Context(contextMap)); Interceptor interceptor = builder.build(); byte[] body = interceptor.intercept( new InterceptorTestHelper.MyEvent("First\002\002Second\002\002\002".getBytes())).getBody(); assertArrayEquals("Six Fields, " + new String(body), "abcSeconddefghi".getBytes(), body); }
@Test public void testInterceptEventWithTerminatingSeparator() { HashMap<String, String> contextMap = new HashMap<String, String>(); contextMap.put(ColumnFilteringInterceptor.Constants.SRC_SEPARATOR, Byte.toString((byte)2)); contextMap.put(ColumnFilteringFormattingInterceptor.Constants.COLUMNS_FORMATTER, "a{1}bc{2}def{3}"); ColumnFilteringFormattingInterceptor.Builder builder = new ColumnFilteringFormattingInterceptor.Builder(); builder.configure(new Context(contextMap)); Interceptor interceptor = builder.build(); byte[] body = interceptor.intercept( new InterceptorTestHelper.MyEvent("First\002\002Second\002\002\002".getBytes())).getBody(); assertArrayEquals("Six Fields, " + new String(body), "abcSeconddef".getBytes(), body); }
@Test public void testInterceptEventWithColumnZero() { HashMap<String, String> contextMap = new HashMap<String, String>(); contextMap.put(ColumnFilteringInterceptor.Constants.SRC_SEPARATOR, Byte.toString((byte)2)); contextMap.put(ColumnFilteringFormattingInterceptor.Constants.COLUMNS_FORMATTER, "{0}\001"); ColumnFilteringFormattingInterceptor.Builder builder = new ColumnFilteringFormattingInterceptor.Builder(); builder.configure(new Context(contextMap)); Interceptor interceptor = builder.build(); assertArrayEquals("Empty Bytes", "\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("".getBytes())).getBody()); assertArrayEquals("One Field", "First\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("First".getBytes())).getBody()); assertArrayEquals("Two Fields", "\001".getBytes(), interceptor.intercept(new InterceptorTestHelper.MyEvent("\002First".getBytes())).getBody()); }