package com.cloudera.flume.handlers.thrift;
import org.junit.Assert;
import org.junit.Test;
import com.cloudera.flume.core.Event;
import com.cloudera.flume.core.EventImpl;
import com.cloudera.flume.handlers.thrift.PrioritizedThriftEventSource.EventQueue;
final Event e1 = new EventImpl("A".getBytes(), Event.Priority.INFO);
final Event e2 = new EventImpl("B".getBytes(), Event.Priority.INFO);
final Event e3 = new EventImpl("C".getBytes(), Event.Priority.INFO);
final Event e4 = new EventImpl("D".getBytes(), Event.Priority.INFO);
final Event e5 = new EventImpl("E".getBytes(), Event.Priority.WARN);
final Event e6 = new EventImpl("F".getBytes(), Event.Priority.DEBUG);
@SuppressWarnings("serial")
@Test
EventQueue q = new EventQueue(10) {
{
add(e5);
add(e6);
add(e1);
}
};
Assert.assertEquals(q.take(), e5);
Assert.assertEquals(q.take(), e1);
Assert.assertEquals(q.take(), e6);
}
@SuppressWarnings("serial")
@Test
EventQueue q = new EventQueue(10) {
{
add(e4);
System.out.println("insert " + e4.getTimestamp() + " " + e4.getNanos()
+ " " + e4);
add(e2);
System.out.println("insert " + e2.getTimestamp() + " " + e2.getNanos()
+ " " + e2);
add(e1);
System.out.println("insert " + e1.getTimestamp() + " " + e1.getNanos()
+ " " + e1);
add(e3);
System.out.println("insert " + e3.getTimestamp() + " " + e3.getNanos()
+ " " + e3);
}
};
Event[] ordered = { e1, e2, e3, e4 };
for (int i = 0; i < ordered.length; i++) {
Event qe = q.take();
System.out.println("take " + qe.getTimestamp() + " " + qe.getNanos()
+ " " + qe);
Assert.assertEquals(ordered[i], qe);
}
}
@SuppressWarnings("serial")
@Test
Thread.sleep(100);
final Event e = new EventImpl("test".getBytes());
EventQueue q = new EventQueue(10) {
{
add(e);
System.out.println("insert " + e.getTimestamp() + " " + e.getNanos()
+ " " + e);
add(e4);
System.out.println("insert " + e4.getTimestamp() + " " + e4.getNanos()
+ " " + e4);
add(e2);
System.out.println("insert " + e2.getTimestamp() + " " + e2.getNanos()
+ " " + e2);
add(e1);
System.out.println("insert " + e1.getTimestamp() + " " + e1.getNanos()
+ " " + e1);
add(e3);
System.out.println("insert " + e3.getTimestamp() + " " + e3.getNanos()
+ " " + e3);
}
};
Event[] ordered = { e1, e2, e3, e4, e };
for (int i = 0; i < ordered.length; i++) {
Event qe = q.take();
System.out.println("take " + qe.getTimestamp() + " " + qe.getNanos()
+ " " + qe);
Assert.assertEquals(ordered[i], qe);
}
}
}