RecoverableMemoryChannel extends BasicChannelSemantics { @Override public void configure(Context context) { memoryChannel.configure(context); int capacity = context.getInteger(CAPACITY, DEFAULT_CAPACITY); if(queueRemaining == null) { queueRemaining = new Semaphore(capacity, true); } else if(capacity > this.capacity) { queueRemaining.release(capacity - this.capacity); } else if(capacity < this.capacity) { queueRemaining.acquireUninterruptibly(this.capacity - capacity); } this.capacity = capacity; keepAlive = context.getInteger(KEEPALIVE, DEFAULT_KEEPALIVE); long rollSize = context.getLong(WAL_ROLL_SIZE, WAL.DEFAULT_ROLL_SIZE); long maxLogsSize = context.getLong(WAL_MAX_LOGS_SIZE, WAL.DEFAULT_MAX_LOGS_SIZE); long minLogRetentionPeriod = context.getLong(WAL_MIN_RENTENTION_PERIOD, WAL.DEFAULT_MIN_LOG_RENTENTION_PERIOD); long workerInterval = context.getLong(WAL_WORKER_INTERVAL, WAL.DEFAULT_WORKER_INTERVAL); if(wal == null) { String homePath = System.getProperty("user.home").replace('\\', '/'); String dataDir = context.getString(WAL_DATA_DIR, homePath + "/.flume/recoverable-memory-channel"); try { wal = new WAL<RecoverableMemoryChannelEvent>(new File(dataDir), RecoverableMemoryChannelEvent.class, rollSize, maxLogsSize, minLogRetentionPeriod, workerInterval); } catch (IOException e) { Throwables.propagate(e); } } else { wal.setRollSize(rollSize); wal.setMaxLogsSize(maxLogsSize); wal.setMinLogRetentionPeriod(minLogRetentionPeriod); wal.setWorkerInterval(workerInterval); LOG.warn(this.getClass().getSimpleName() + " only supports " + "partial reconfiguration."); } } 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 testReconfigure() 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()); } Configurables.configure(channel, context); List<String> out = takeEvents(channel, 1, Integer.MAX_VALUE); Collections.sort(in); Collections.sort(out); Assert.assertEquals(in, out); }