FileChannel extends BasicChannelSemantics { @Override public void configure(Context context) { String homePath = System.getProperty("user.home").replace('\\', '/'); String strCheckpointDir = context.getString(FileChannelConfiguration.CHECKPOINT_DIR, homePath + "/.flume/file-channel/checkpoint"); String[] strDataDirs = context.getString(FileChannelConfiguration.DATA_DIRS, homePath + "/.flume/file-channel/data").split(","); if(checkpointDir == null) { checkpointDir = new File(strCheckpointDir); } else if(!checkpointDir.getAbsolutePath(). equals(new File(strCheckpointDir).getAbsolutePath())) { LOG.warn("An attempt was made to change the checkpoint " + "directory after start, this is not supported."); } if(dataDirs == null) { dataDirs = new File[strDataDirs.length]; for (int i = 0; i < strDataDirs.length; i++) { dataDirs[i] = new File(strDataDirs[i]); } } else { boolean changed = false; if(dataDirs.length != strDataDirs.length) { changed = true; } else { for (int i = 0; i < strDataDirs.length; i++) { if(!dataDirs[i].getAbsolutePath(). equals(new File(strDataDirs[i]).getAbsolutePath())) { changed = true; break; } } } if(changed) { LOG.warn("An attempt was made to change the data " + "directories after start, this is not supported."); } } int newCapacity = context.getInteger(FileChannelConfiguration.CAPACITY, FileChannelConfiguration.DEFAULT_CAPACITY); if(capacity > 0 && newCapacity != capacity) { LOG.warn("Capacity of this channel cannot be sized on the fly due " + "the requirement we have enough DirectMemory for the queue and " + "downsizing of the queue cannot be guranteed due to the " + "fact there maybe more items on the queue than the new capacity."); } else { capacity = newCapacity; } keepAlive = context.getInteger(FileChannelConfiguration.KEEP_ALIVE, FileChannelConfiguration.DEFAULT_KEEP_ALIVE); transactionCapacity = context.getInteger(FileChannelConfiguration.TRANSACTION_CAPACITY, FileChannelConfiguration.DEFAULT_TRANSACTION_CAPACITY); checkpointInterval = context.getLong(FileChannelConfiguration.CHECKPOINT_INTERVAL, FileChannelConfiguration.DEFAULT_CHECKPOINT_INTERVAL); maxFileSize = Math.min( context.getLong(FileChannelConfiguration.MAX_FILE_SIZE, FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE), FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE); if(queueRemaining == null) { queueRemaining = new Semaphore(capacity, true); } if(log != null) { log.setCheckpointInterval(checkpointInterval); log.setMaxFileSize(maxFileSize); } }  @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop();  }
@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); }