package com.cloudera.flume.core;
import static org.junit.Assert.assertEquals;
import java.io.IOException;
import org.junit.Test;
import com.cloudera.flume.conf.Context;
import com.cloudera.flume.conf.FlumeSpecException;
import com.cloudera.flume.handlers.debug.MemorySinkSource;
@Test
MemorySinkSource mem = new MemorySinkSource();
EventSink sel = new SelectDecorator<EventSink>(mem, "foo", "bar", "bork");
Event e = new EventImpl("content".getBytes());
Attributes.setString(e, "foo", "foo data");
Attributes.setString(e, "bar", "bar data");
Attributes.setString(e, "baz", "baz data");
sel.open();
sel.append(e);
sel.close();
mem.open();
Event e2 = mem.next();
assertEquals("foo data", Attributes.readString(e2, "foo"));
assertEquals("bar data", Attributes.readString(e2, "bar"));
assertEquals(null, Attributes.readString(e2, "baz"));
assertEquals(null, Attributes.readString(e2, "bork"));
}
@Test
EventSink snk =
new CompositeSink(new Context(),
"{ select(\"foo\", \"bar\",\"bork\") => counter(\"count\") }");
snk.open();
Event e = new EventImpl("content".getBytes());
Attributes.setString(e, "foo", "foo data");
Attributes.setString(e, "bar", "bar data");
Attributes.setString(e, "baz", "baz data");
snk.append(e);
snk.close();
}
@Test
public void testMask()
throws IOException {
MemorySinkSource mem = new MemorySinkSource();
EventSink mask = new MaskDecorator<EventSink>(mem, "foo", "bar", "bork");
Event e = new EventImpl("content".getBytes());
Attributes.setString(e, "foo", "foo data");
Attributes.setString(e, "bar", "bar data");
Attributes.setString(e, "baz", "baz data");
mask.open();
mask.append(e);
mask.close();
mem.open();
Event e2 = mem.next();
assertEquals(null, Attributes.readString(e2, "foo"));
assertEquals(null, Attributes.readString(e2, "bar"));
assertEquals("baz data", Attributes.readString(e2, "baz"));
assertEquals(null, Attributes.readString(e2, "bork"));
}
@Test
EventSink snk =
new CompositeSink(new Context(),
"{ mask(\"foo\", \"bar\",\"bork\") => counter(\"count\") }");
snk.open();
Event e = new EventImpl("content".getBytes());
Attributes.setString(e, "foo", "foo data");
Attributes.setString(e, "bar", "bar data");
Attributes.setString(e, "baz", "baz data");
snk.append(e);
snk.close();
}
}