DigestDecorator extends EventSinkDecorator<S> { @Override public synchronized void append(Event e) throws IOException, InterruptedException { this.stomach.reset(); this.stomach.update(e.getBody()); byte digest[] = this.stomach.digest(); if (this.encodeBase64) { digest = Base64.encodeBase64(digest); } e.set(this.attr, digest); super.append(e); } DigestDecorator(S s, String algorithm, String attr, boolean encodeBase64); @Override synchronized void append(Event e); static SinkDecoBuilder builder();  }
@Test public void testDigestBuilder() throws IOException, FlumeSpecException, InterruptedException { EventSink snk = new CompositeSink(new Context(), "{ digest(\"MD5\", \"digest\") => counter(\"count\") }"); snk.open(); Event e = new EventImpl("content".getBytes()); snk.append(e); snk.close(); }
@Test public void testDigestBuilderBase64() throws IOException, FlumeSpecException, InterruptedException { EventSink snk = new CompositeSink(new Context(), "{ digest(\"MD5\", \"digest\", base64=\"true\") => counter(\"count\") }"); snk.open(); Event e = new EventImpl("content".getBytes()); snk.append(e); snk.close(); }