RecoverableMemoryChannel extends BasicChannelSemantics { @Override public synchronized void stop() { open = false; LOG.info("Stopping " + this); try { close(); } catch (IOException e) { Throwables.propagate(e); } super.stop(); } RecoverableMemoryChannel(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); static final String WAL_DATA_DIR; static final String WAL_ROLL_SIZE; static final String WAL_MAX_LOGS_SIZE; static final String WAL_MIN_RENTENTION_PERIOD; static final String WAL_WORKER_INTERVAL; static final String CAPACITY; static final String KEEPALIVE; static final int DEFAULT_CAPACITY; static final int DEFAULT_KEEPALIVE; }
@Test public void testRestart() throws Exception { List<String> in = Lists.newArrayList(); try { while(true) { in.addAll(putEvents(channel, "restart", 1, 1)); } } catch (ChannelException e) { Assert.assertEquals("Cannot acquire capacity", e.getMessage()); } channel.stop(); channel = createFileChannel(); List<String> out = takeEvents(channel, 1, Integer.MAX_VALUE); Collections.sort(in); Collections.sort(out); Assert.assertEquals(in, out); }