// Load the source configuration JSON file
//File sourcesJson = new File("integration-test/config/sources-member2.json");
File sourcesJson = new File(_sSourcesConfigFile);
ObjectMapper mapper = new ObjectMapper();
PhysicalSourceConfig physicalSourceConfig = mapper.readValue(sourcesJson, PhysicalSourceConfig.class);
physicalSourceConfig.checkForNulls();
Config config = new Config();
ConfigLoader<StaticConfig> configLoader =
new ConfigLoader<StaticConfig>("databus.seed.", config);
_sStaticConfig = configLoader.loadConfig(_sBootstrapConfigProps);
// Make sure the URI from the configuration file identifies an Oracle JDBC source.
String uri = physicalSourceConfig.getUri();
if(!uri.startsWith("jdbc:oracle"))
{
throw new InvalidConfigException("Invalid source URI (" +
physicalSourceConfig.getUri() + "). Only jdbc:oracle: URIs are supported.");
}
String sourceTypeStr = physicalSourceConfig.getReplBitSetter().getSourceType();
if (SourceType.TOKEN.toString().equalsIgnoreCase(sourceTypeStr))
throw new InvalidConfigException("Token Source-type for Replication bit setter config cannot be set for trigger-based Databus relay !!");
// Create the OracleDataSource used to get DB connection(s)
try
{
Class oracleDataSourceClass = OracleJarUtils.loadClass("oracle.jdbc.pool.OracleDataSource");
Object ods = oracleDataSourceClass.newInstance();
Method setURLMethod = oracleDataSourceClass.getMethod("setURL", String.class);
setURLMethod.invoke(ods, uri);
_sDataStore = (DataSource) ods;
} catch (Exception e)
{
String errMsg = "Error creating a data source object ";
LOG.error(errMsg, e);
throw e;
}
//TODO: Need a better way than relaying on RelayFactory for generating MonitoredSourceInfo
OracleEventProducerFactory factory = new BootstrapSeederOracleEventProducerFactory(_sStaticConfig.getController().getPKeyNameMap());
// Parse each one of the logical sources
_sources = new ArrayList<OracleTriggerMonitoredSourceInfo>();
FileSystemSchemaRegistryService schemaRegistryService =
FileSystemSchemaRegistryService.build(_sStaticConfig.getSchemaRegistry().getFileSystem());
Set<String> seenUris = new HashSet<String>();
for(LogicalSourceConfig sourceConfig : physicalSourceConfig.getSources())
{
String srcUri = sourceConfig.getUri();
if ( seenUris.contains(srcUri))
{
String msg = "Uri (" + srcUri + ") is used for more than one sources. Currently Bootstrap Seeder cannot support seeding sources with the same URI together. Please have them run seperately !!";
LOG.fatal(msg);
throw new InvalidConfigException(msg);
}
seenUris.add(srcUri);
OracleTriggerMonitoredSourceInfo source =
factory.buildOracleMonitoredSourceInfo(sourceConfig.build(), physicalSourceConfig.build(), schemaRegistryService);
_sources.add(source);
}
_sSeeder = new BootstrapDBSeeder(_sStaticConfig.getBootstrap(),_sources);
_sBootstrapBuffer = new BootstrapEventBuffer(_sStaticConfig.getController().getCommitInterval() * 2);