@Test(expected = MongoSinkException.class) public void confSingleModeWithNoDefaultDBCollection() throws Exception { final MongoSink mongoSink = new MongoSink(); final Context context = new Context(); context.put("dynamic", "false"); context.put("mongoUri", "mongodb: Configurables.configure(mongoSink, context); }
@Override public void configure(Context context) { try { if (!"INJECTED".equals(context.getString(CONF_URI))) { this.mongoClientURI = new MongoClientURI( context.getString(CONF_URI), MongoClientOptions.builder().writeConcern(WriteConcern.SAFE) ); this.mongoClient = new MongoClient(mongoClientURI); if (mongoClientURI.getDatabase() != null) { this.mongoDefaultDb = mongoClient.getDB(mongoClientURI.getDatabase()); } if (mongoClientURI.getCollection() != null) { this.mongoDefaultCollection = mongoDefaultDb.getCollection(mongoClientURI.getCollection()); } } final String mappingFilename = context.getString(CONF_MAPPING_FILE); this.eventParser = (mappingFilename == null) ? new EventParser() : new EventParser(MappingDefinition.load(mappingFilename)); this.isDynamicMode = context.getBoolean(CONF_DYNAMIC, DEFAULT_DYNAMIC); if (!isDynamicMode && mongoDefaultCollection == null) { throw new MongoSinkException("Default MongoDB collection must be specified unless dynamic mode is enabled"); } this.dynamicDBField = context.getString(CONF_DYNAMIC_DB_FIELD, DEFAULT_DYNAMIC_DB_FIELD); this.dynamicCollectionField = context.getString(CONF_DYNAMIC_COLLECTION_FIELD, DEFAULT_DYNAMIC_COLLECTION_FIELD); this.sinkCounter = new SinkCounter(this.getName()); this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.updateInsteadReplace = context.getBoolean(CONF_UPDATE_INSTEAD_REPLACE,DEFAULT_UPDATE_INSTEAD_REPLACE); } catch (IOException ex) { throw new MongoSinkException(ex); } }
MongoSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { try { if (!"INJECTED".equals(context.getString(CONF_URI))) { this.mongoClientURI = new MongoClientURI( context.getString(CONF_URI), MongoClientOptions.builder().writeConcern(WriteConcern.SAFE) ); this.mongoClient = new MongoClient(mongoClientURI); if (mongoClientURI.getDatabase() != null) { this.mongoDefaultDb = mongoClient.getDB(mongoClientURI.getDatabase()); } if (mongoClientURI.getCollection() != null) { this.mongoDefaultCollection = mongoDefaultDb.getCollection(mongoClientURI.getCollection()); } } final String mappingFilename = context.getString(CONF_MAPPING_FILE); this.eventParser = (mappingFilename == null) ? new EventParser() : new EventParser(MappingDefinition.load(mappingFilename)); this.isDynamicMode = context.getBoolean(CONF_DYNAMIC, DEFAULT_DYNAMIC); if (!isDynamicMode && mongoDefaultCollection == null) { throw new MongoSinkException("Default MongoDB collection must be specified unless dynamic mode is enabled"); } this.dynamicDBField = context.getString(CONF_DYNAMIC_DB_FIELD, DEFAULT_DYNAMIC_DB_FIELD); this.dynamicCollectionField = context.getString(CONF_DYNAMIC_COLLECTION_FIELD, DEFAULT_DYNAMIC_COLLECTION_FIELD); this.sinkCounter = new SinkCounter(this.getName()); this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.updateInsteadReplace = context.getBoolean(CONF_UPDATE_INSTEAD_REPLACE,DEFAULT_UPDATE_INSTEAD_REPLACE); } catch (IOException ex) { throw new MongoSinkException(ex); } } }
MongoSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { try { if (!"INJECTED".equals(context.getString(CONF_URI))) { this.mongoClientURI = new MongoClientURI( context.getString(CONF_URI), MongoClientOptions.builder().writeConcern(WriteConcern.SAFE) ); this.mongoClient = new MongoClient(mongoClientURI); if (mongoClientURI.getDatabase() != null) { this.mongoDefaultDb = mongoClient.getDB(mongoClientURI.getDatabase()); } if (mongoClientURI.getCollection() != null) { this.mongoDefaultCollection = mongoDefaultDb.getCollection(mongoClientURI.getCollection()); } } final String mappingFilename = context.getString(CONF_MAPPING_FILE); this.eventParser = (mappingFilename == null) ? new EventParser() : new EventParser(MappingDefinition.load(mappingFilename)); this.isDynamicMode = context.getBoolean(CONF_DYNAMIC, DEFAULT_DYNAMIC); if (!isDynamicMode && mongoDefaultCollection == null) { throw new MongoSinkException("Default MongoDB collection must be specified unless dynamic mode is enabled"); } this.dynamicDBField = context.getString(CONF_DYNAMIC_DB_FIELD, DEFAULT_DYNAMIC_DB_FIELD); this.dynamicCollectionField = context.getString(CONF_DYNAMIC_COLLECTION_FIELD, DEFAULT_DYNAMIC_COLLECTION_FIELD); this.sinkCounter = new SinkCounter(this.getName()); this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.updateInsteadReplace = context.getBoolean(CONF_UPDATE_INSTEAD_REPLACE,DEFAULT_UPDATE_INSTEAD_REPLACE); } catch (IOException ex) { throw new MongoSinkException(ex); } } MongoSink(); }
MongoSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { try { if (!"INJECTED".equals(context.getString(CONF_URI))) { this.mongoClientURI = new MongoClientURI( context.getString(CONF_URI), MongoClientOptions.builder().writeConcern(WriteConcern.SAFE) ); this.mongoClient = new MongoClient(mongoClientURI); if (mongoClientURI.getDatabase() != null) { this.mongoDefaultDb = mongoClient.getDB(mongoClientURI.getDatabase()); } if (mongoClientURI.getCollection() != null) { this.mongoDefaultCollection = mongoDefaultDb.getCollection(mongoClientURI.getCollection()); } } final String mappingFilename = context.getString(CONF_MAPPING_FILE); this.eventParser = (mappingFilename == null) ? new EventParser() : new EventParser(MappingDefinition.load(mappingFilename)); this.isDynamicMode = context.getBoolean(CONF_DYNAMIC, DEFAULT_DYNAMIC); if (!isDynamicMode && mongoDefaultCollection == null) { throw new MongoSinkException("Default MongoDB collection must be specified unless dynamic mode is enabled"); } this.dynamicDBField = context.getString(CONF_DYNAMIC_DB_FIELD, DEFAULT_DYNAMIC_DB_FIELD); this.dynamicCollectionField = context.getString(CONF_DYNAMIC_COLLECTION_FIELD, DEFAULT_DYNAMIC_COLLECTION_FIELD); this.sinkCounter = new SinkCounter(this.getName()); this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.updateInsteadReplace = context.getBoolean(CONF_UPDATE_INSTEAD_REPLACE,DEFAULT_UPDATE_INSTEAD_REPLACE); } catch (IOException ex) { throw new MongoSinkException(ex); } } MongoSink(); @Override void configure(Context context); @Override Status process(); @Override synchronized void start(); @Override synchronized void stop(); }
MongoSink extends AbstractSink implements Configurable { @Override public void configure(Context context) { try { if (!"INJECTED".equals(context.getString(CONF_URI))) { this.mongoClientURI = new MongoClientURI( context.getString(CONF_URI), MongoClientOptions.builder().writeConcern(WriteConcern.SAFE) ); this.mongoClient = new MongoClient(mongoClientURI); if (mongoClientURI.getDatabase() != null) { this.mongoDefaultDb = mongoClient.getDB(mongoClientURI.getDatabase()); } if (mongoClientURI.getCollection() != null) { this.mongoDefaultCollection = mongoDefaultDb.getCollection(mongoClientURI.getCollection()); } } final String mappingFilename = context.getString(CONF_MAPPING_FILE); this.eventParser = (mappingFilename == null) ? new EventParser() : new EventParser(MappingDefinition.load(mappingFilename)); this.isDynamicMode = context.getBoolean(CONF_DYNAMIC, DEFAULT_DYNAMIC); if (!isDynamicMode && mongoDefaultCollection == null) { throw new MongoSinkException("Default MongoDB collection must be specified unless dynamic mode is enabled"); } this.dynamicDBField = context.getString(CONF_DYNAMIC_DB_FIELD, DEFAULT_DYNAMIC_DB_FIELD); this.dynamicCollectionField = context.getString(CONF_DYNAMIC_COLLECTION_FIELD, DEFAULT_DYNAMIC_COLLECTION_FIELD); this.sinkCounter = new SinkCounter(this.getName()); this.batchSize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE); this.updateInsteadReplace = context.getBoolean(CONF_UPDATE_INSTEAD_REPLACE,DEFAULT_UPDATE_INSTEAD_REPLACE); } catch (IOException ex) { throw new MongoSinkException(ex); } } MongoSink(); @Override void configure(Context context); @Override Status process(); @Override synchronized void start(); @Override synchronized void stop();  }