import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
@Override
public void configure(final Env env, final Config conf, final Binder binder) {
  Key<DataSource> dskey = Key.get(DataSource.class, Names.named(name));
  HikariDataSource ds = (HikariDataSource) env.get(dskey)
      .orElseThrow(() -> new NoSuchElementException("DataSource missing: " + dskey));
  Configuration jooqconf = new DefaultConfiguration();
  ConnectionProvider dscp = new DataSourceConnectionProvider(ds);
  jooqconf.set(JDBCUtils.dialect(ds.getDataSourceProperties().getProperty("url")));
  jooqconf.set(dscp);
  jooqconf.set(new DefaultTransactionProvider(dscp));

  if (callback != null) {
    callback.accept(jooqconf, conf);
  }

  ServiceKey serviceKey = env.serviceKey();
  serviceKey.generate(Configuration.class, name, k -> binder.bind(k).toInstance(jooqconf));

  Provider<DSLContext> dsl = () -> DSL.using(jooqconf);
  serviceKey.generate(DSLContext.class, name, k -> binder.bind(k).toProvider(dsl));
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
DSLContext create(final Connection connection) {
  try {
    final SQLDialect dialect = JDBCUtils.dialect(connection.getMetaData().getURL());

    if (dialect == SQLDialect.HSQLDB) {
      return hsqldb(connection);
    }

    return defaultContext(connection);
  } catch (final SQLException e) {
    return defaultContext(connection);
  }
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
@SuppressWarnings( "deprecation" )
@Override
   public void configure(Context context) {
   	// DBCP 초기화
   	ConnectionManager.instance.initialize( context );

       this.batchsize = context.getInteger(CONF_BATCH_SIZE, DEFAULT_BATCH_SIZE);

       this.sqlDialect = SQLDialect.valueOf(context.getString(CONF_SQL_DIALECT).toUpperCase(Locale.ENGLISH));

       final String sql = context.getString(CONF_SQL);
       if (sql == null) {
       	Connection connection = null;
           try {
           	// Table 정보 매핑
           	connection = ConnectionManager.instance.getConnection();
           	final DSLContext create = DSL.using(connection, sqlDialect);
           	this.queryGenerator = new MappingQueryGenerator(create, context.getString(CONF_TABLE));
           } catch (SQLException ex) {
               throw new JDBCSinkException(ex);
           } finally {
           	JDBCUtils.safeClose( connection );
           }
       } else {
           this.queryGenerator = new TemplateQueryGenerator(sqlDialect, sql);
       }

       this.sinkCounter = new SinkCounter(this.getName());
   }
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
@Override
public final void migrate(Connection connection) throws Exception {
    didExecute = true;
    SQLDialect dialect = JDBCUtils.dialect(connection);
    if (SQLDialect.DEFAULT.equals(dialect))
        throw new IllegalStateException("Dialect couldn't be deducted from connection " + connection);

    Configuration configuration = new DefaultConfiguration().set(connection).set(dialect);
    DdlExecuteListener listener = new DdlExecuteListener();
    configuration.set(new DefaultExecuteListenerProvider(listener));
    DSLContext create = DSL.using(configuration);

    migrate(connection, create);
    ddlInstructionExecuted = listener.ddlInstructionExecuted();
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
DSLContext get(Connection conn) {
  return DSL.using(new DefaultConfiguration()
      .set(conn)
      .set(JDBCUtils.dialect(conn))
      .set(settings)
      .set(listenerProvider));
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
private SQLDialect determineDialect(PooledDataSourceFactory dataSourceFactory, ManagedDataSource dataSource) {
    // If a dialect was specified, great!
    if (getDialect().isPresent()) {
        return dialect.get();
    }

    return JDBCUtils.dialect(dataSourceFactory.getUrl());
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
@Override
public Status process() throws EventDeliveryException {
    Status status = Status.BACKOFF;
    Transaction transaction = this.getChannel().getTransaction();
    Connection connection = null;

    try {
    	transaction.begin();

    	connection = ConnectionManager.instance.getConnection();
    	final DSLContext create = DSL.using(connection, sqlDialect);

    	List<Event> eventList = this.takeEventsFromChannel( this.getChannel(), this.batchsize);
        status = Status.READY;
        if (!eventList.isEmpty()) {
            if (eventList.size() == this.batchsize) {
                this.sinkCounter.incrementBatchCompleteCount();
            } else {
                this.sinkCounter.incrementBatchUnderflowCount();
            }

            final boolean success = this.queryGenerator.executeQuery(create, eventList);

            if (!success) {
                throw new JDBCSinkException("Query failed");
            }

            connection.commit();

            this.sinkCounter.addToEventDrainSuccessCount(eventList.size());
        } else {
            this.sinkCounter.incrementBatchEmptyCount();
        }

        transaction.commit();
        status = Status.READY;
    } catch (Throwable t) {
        log.error("Exception during process", t);
        try {
        	connection.rollback();
        } catch (Exception ex) {
            log.error("Exception on rollback", ex);
        } finally {
            transaction.rollback();
            status = Status.BACKOFF;
            this.sinkCounter.incrementConnectionFailedCount();
            if (t instanceof Error) {
                throw new JDBCSinkException(t);
            }
        }
    } finally {
        transaction.close();
        JDBCUtils.safeClose( connection );
    }
    return status;
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
private static SQLDialect extractDialectOrThrow(Connection connection) {
    SQLDialect dialect = JDBCUtils.dialect(connection);
    if (SQLDialect.DEFAULT.equals(dialect))
        throw new IllegalStateException("Dialect couldn't be deducted from connection " + connection);
    return dialect;
}
 

import org.jooq.tools.jdbc.JDBCUtils; //导入依赖的package包/类
protected DSLContext createDSLContext()
{
  return DSL.using(store.getConnection(), JDBCUtils.dialect(store.getDatabaseUrl()));
}