svn commit: r1156890 - /incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java

[es...@apache.org]

Author: esammer
Date: Fri Aug 12 00:47:24 2011
New Revision: 1156890

URL: http://svn.apache.org/viewvc?rev=1156890&view=rev
Log:
- Added a test of an interrupted open(). - Fixed a bad import of junit Assert.

Modified:
    incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java

Modified: incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java?rev=1156890&r1=1156889&r2=1156890&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/core/TestChannelDriver.java Fri Aug 12 00:47:24 2011
@@ -1,12 +1,11 @@
 package org.apache.flume.core;
 
-import junit.framework.Assert;
-
 import org.apache.flume.lifecycle.LifecycleController;
 import org.apache.flume.lifecycle.LifecycleException;
 import org.apache.flume.lifecycle.LifecycleState;
 import org.apache.flume.sink.NullSink;
 import org.apache.flume.source.SequenceGeneratorSource;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -27,6 +26,7 @@ public class TestChannelDriver {
   @Test
   public void testNormalLifecycle() throws LifecycleException,
       InterruptedException {
+
     final EventSource source = new SequenceGeneratorSource();
     final EventSink sink = new NullSink();
     final CounterGroup sourceCounters = new CounterGroup();
@@ -128,6 +128,111 @@ public class TestChannelDriver {
   }
 
   @Test
+  public void testInterruptOpen() throws LifecycleException,
+      InterruptedException {
+
+    final EventSource source = new SequenceGeneratorSource();
+    final EventSink sink = new NullSink();
+    final CounterGroup sourceCounters = new CounterGroup();
+    final CounterGroup sinkCounters = new CounterGroup();
+
+    sourceCounters.setName("source");
+    sinkCounters.setName("sink");
+
+    driver.setSource(new EventSource() {
+
+      @Override
+      public Event<?> next(Context context) throws InterruptedException,
+          MessageDeliveryException {
+
+        sourceCounters.incrementAndGet("next");
+
+        return source.next(context);
+      }
+
+      @Override
+      public void open(Context context) throws InterruptedException,
+          LifecycleException {
+
+        sourceCounters.incrementAndGet("open");
+
+        source.open(context);
+      }
+
+      @Override
+      public void close(Context context) throws InterruptedException,
+          LifecycleException {
+
+        sourceCounters.incrementAndGet("close");
+
+        source.open(context);
+      }
+    });
+
+    driver.setSink(new EventSink() {
+
+      @Override
+      public void append(Context context, Event<?> event)
+          throws InterruptedException, MessageDeliveryException {
+
+        sinkCounters.incrementAndGet("append");
+
+        sink.append(context, event);
+      }
+
+      @Override
+      public void open(Context context) throws InterruptedException,
+          LifecycleException {
+
+        sinkCounters.incrementAndGet("open");
+
+        sink.open(context);
+      }
+
+      @Override
+      public void close(Context context) throws InterruptedException,
+          LifecycleException {
+
+        sinkCounters.incrementAndGet("close");
+
+        sink.close(context);
+      }
+    });
+
+    Context context = new Context();
+
+    Thread.currentThread().interrupt();
+    driver.start(context);
+
+    LifecycleController.waitForOneOf(driver, new LifecycleState[] {
+        LifecycleState.START, LifecycleState.ERROR }, 5000);
+
+    Assert.assertEquals(LifecycleState.ERROR, driver.getLifecycleState());
+
+    Thread.sleep(500);
+
+    driver.stop(context);
+
+    LifecycleController.waitForOneOf(driver, new LifecycleState[] {
+        LifecycleState.STOP, LifecycleState.ERROR }, 5000);
+
+    Assert.assertEquals(LifecycleState.STOP, driver.getLifecycleState());
+
+    Assert.assertEquals(Long.valueOf(1), sourceCounters.get("open"));
+    Assert.assertEquals(Long.valueOf(1), sourceCounters.get("close"));
+    Assert.assertTrue(sourceCounters.get("next") > 0);
+    Assert.assertEquals(Long.valueOf(1), sinkCounters.get("open"));
+    Assert.assertEquals(Long.valueOf(1), sinkCounters.get("close"));
+    Assert.assertTrue(sinkCounters.get("append") > 0);
+    Assert.assertEquals(
+        "Source next() events didn't match sink append() events",
+        sourceCounters.get("next"), sinkCounters.get("append"));
+
+    logger.info("Source counters:{} Sink counters:{}", sourceCounters,
+        sinkCounters);
+  }
+
+  @Test
   public void testFailedOpen() throws LifecycleException, InterruptedException {
     final EventSource source = new SequenceGeneratorSource();
     final EventSink sink = new NullSink();