SpoolDirectorySource extends AbstractSource implements Configurable, EventDrivenSource, BatchSizeSupported { @Override public synchronized void configure(Context context) { spoolDirectory = context.getString(SPOOL_DIRECTORY); Preconditions.checkState(spoolDirectory != null, "Configuration must specify a spooling directory"); completedSuffix = context.getString(SPOOLED_FILE_SUFFIX, DEFAULT_SPOOLED_FILE_SUFFIX); deletePolicy = context.getString(DELETE_POLICY, DEFAULT_DELETE_POLICY); fileHeader = context.getBoolean(FILENAME_HEADER, DEFAULT_FILE_HEADER); fileHeaderKey = context.getString(FILENAME_HEADER_KEY, DEFAULT_FILENAME_HEADER_KEY); basenameHeader = context.getBoolean(BASENAME_HEADER, DEFAULT_BASENAME_HEADER); basenameHeaderKey = context.getString(BASENAME_HEADER_KEY, DEFAULT_BASENAME_HEADER_KEY); batchSize = context.getInteger(BATCH_SIZE, DEFAULT_BATCH_SIZE); inputCharset = context.getString(INPUT_CHARSET, DEFAULT_INPUT_CHARSET); decodeErrorPolicy = DecodeErrorPolicy.valueOf( context.getString(DECODE_ERROR_POLICY, DEFAULT_DECODE_ERROR_POLICY) .toUpperCase(Locale.ENGLISH)); includePattern = context.getString(INCLUDE_PAT, DEFAULT_INCLUDE_PAT); ignorePattern = context.getString(IGNORE_PAT, DEFAULT_IGNORE_PAT); trackerDirPath = context.getString(TRACKER_DIR, DEFAULT_TRACKER_DIR); deserializerType = context.getString(DESERIALIZER, DEFAULT_DESERIALIZER); deserializerContext = new Context(context.getSubProperties(DESERIALIZER + ".")); consumeOrder = ConsumeOrder.valueOf(context.getString(CONSUME_ORDER, DEFAULT_CONSUME_ORDER.toString()).toUpperCase(Locale.ENGLISH)); pollDelay = context.getInteger(POLL_DELAY, DEFAULT_POLL_DELAY); recursiveDirectorySearch = context.getBoolean(RECURSIVE_DIRECTORY_SEARCH, DEFAULT_RECURSIVE_DIRECTORY_SEARCH); Integer bufferMaxLineLength = context.getInteger(BUFFER_MAX_LINE_LENGTH); if (bufferMaxLineLength != null && deserializerType != null && deserializerType.equalsIgnoreCase(DEFAULT_DESERIALIZER)) { deserializerContext.put(LineDeserializer.MAXLINE_KEY, bufferMaxLineLength.toString()); } maxBackoff = context.getInteger(MAX_BACKOFF, DEFAULT_MAX_BACKOFF); if (sourceCounter == null) { sourceCounter = new SourceCounter(getName()); } trackingPolicy = context.getString(TRACKING_POLICY, DEFAULT_TRACKING_POLICY); }  @Override synchronized void start(); @Override synchronized void stop(); @Override String toString(); @Override synchronized void configure(Context context); @Override long getBatchSize();  }
@Test(expected = IllegalArgumentException.class) public void testInvalidSortOrder() { Context context = new Context(); context.put(SpoolDirectorySourceConfigurationConstants.SPOOL_DIRECTORY, tmpDir.getAbsolutePath()); context.put(SpoolDirectorySourceConfigurationConstants.CONSUME_ORDER, "undefined"); Configurables.configure(source, context); }
@Test public void testValidSortOrder() { Context context = new Context(); context.put(SpoolDirectorySourceConfigurationConstants.SPOOL_DIRECTORY, tmpDir.getAbsolutePath()); context.put(SpoolDirectorySourceConfigurationConstants.CONSUME_ORDER, "oLdESt"); Configurables.configure(source, context); context.put(SpoolDirectorySourceConfigurationConstants.CONSUME_ORDER, "yoUnGest"); Configurables.configure(source, context); context.put(SpoolDirectorySourceConfigurationConstants.CONSUME_ORDER, "rAnDom"); Configurables.configure(source, context); }