import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
/**
 *
 */
@OnEnabled
public void enable(ConfigurationContext context) throws InitializationException {
    try {
        if (!this.configured) {
            if (logger.isInfoEnabled()) {
                logger.info("Configuring " + this.getClass().getSimpleName() + " for '"
                        + context.getProperty(CONNECTION_FACTORY_IMPL).evaluateAttributeExpressions().getValue() + "' to be connected to '"
                        + context.getProperty(BROKER_URI).evaluateAttributeExpressions().getValue() + "'");
            }
            // will load user provided libraries/resources on the classpath
            Utils.addResourcesToClasspath(context.getProperty(CLIENT_LIB_DIR_PATH).evaluateAttributeExpressions().getValue());

            this.createConnectionFactoryInstance(context);
            if(!isSolace(context))
                this.setConnectionFactoryProperties(context);
        }
        this.configured = true;
    } catch (Exception e) {
        logger.error("Failed to configure " + this.getClass().getSimpleName(), e);
        this.configured = false;
        throw new IllegalStateException(e);
    }
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
/**
 *
 */
@OnEnabled
public void enable(ConfigurationContext context) throws InitializationException {
    try {
        if (!this.configured) {
            if (logger.isInfoEnabled()) {
                logger.info("Configuring " + this.getClass().getSimpleName() + " for '"
                        + context.getProperty(CONNECTION_FACTORY_IMPL).evaluateAttributeExpressions().getValue() + "' to be connected to '"
                        + BROKER_URI + "'");
            }
            // will load user provided libraries/resources on the classpath
            Utils.addResourcesToClasspath(context.getProperty(CLIENT_LIB_DIR_PATH).evaluateAttributeExpressions().getValue());

            this.createConnectionFactoryInstance(context);

            this.setConnectionFactoryProperties(context);
        }
        this.configured = true;
    } catch (Exception e) {
        logger.error("Failed to configure " + this.getClass().getSimpleName(), e);
        this.configured = false;
        throw new IllegalStateException(e);
    }
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
/**
 * @param context the configuration context
 * @throws InitializationException if unable to create a database connection
 */
@OnEnabled
public void onEnabled(final ConfigurationContext context) throws InitializationException {
    Set<AuthorizationToken> tokensToUse = new HashSet<>();

    final String authorizationFilePath = context.getProperty(AUTHORIZATION_FILE_PATH).getValue();

    try (Scanner authsScanner = new Scanner(new File(authorizationFilePath))) {
        authsScanner.useDelimiter(",");
        while (authsScanner.hasNext()) {
            tokensToUse.add(new StandardAuthorizationToken(authsScanner.next()));
        }
    } catch (IOException ioe) {
        throw new InitializationException("Could not initialize " + this.getClass().getSimpleName() + " when trying to acquire authorizations.", ioe);
    }
    getLogger().info("Providing authorities for the following: {}", new Object[]{tokensToUse});
    this.authorizationTokens = tokensToUse;
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
@OnEnabled
public void initialize(final ConfigurationContext context) {
    if (cache == null) {
        final int maxSize = context.getProperty(MAX_SIZE).asInteger();
        final Long expireDuration = context.getProperty(AGE_OFF_DURATION).asTimePeriod(TimeUnit.MILLISECONDS);
        final int concurrencyLevel = context.getProperty(CONCURRENCY_LEVEL).isSet() ?
                context.getProperty(CONCURRENCY_LEVEL).asInteger() : -1;
        cache = new SetCacheImpl<>(maxSize, expireDuration, TimeUnit.MILLISECONDS, concurrencyLevel);
    }
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
@OnEnabled
public void startServer(final ConfigurationContext context)  throws InitializationException {
    vertx = Vertx.vertx();
    eventBus = vertx.eventBus();
    getLogger().info("initializing vertx server...");
    createSockJSServer(context);
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
@OnEnabled
public void start(final ConfigurationContext context) throws IOException {
    createCache(context);

    startServer(context);
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
@OnEnabled
public void initialize(final ConfigurationContext context) {
    this.closed = false;
    this.configContext = context;
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
/**
 * Configures connection pool by creating an instance of the
 * {@link BasicDataSource} based on configuration provided with
 * {@link ConfigurationContext}.
 *
 * This operation makes no guarantees that the actual connection could be
 * made since the underlying system may still go off-line during normal
 * operation of the connection pool.
 *
 * @param context
 *            the configuration context
 * @throws InitializationException
 *             if unable to create a database connection
 */
@OnEnabled
public void onEnabled(final ConfigurationContext context) throws InitializationException {

    final String dburl = context.getProperty(DATABASE_URL).getValue();
    final Integer maxTotal = context.getProperty(MAX_TOTAL_SESSIONS).asInteger();
    final String user = context.getProperty(DB_USER).getValue();
    final String passw = context.getProperty(DB_PASSWORD).getValue();
    // final Long maxWaitMillis = context.getProperty(MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS);

    driver = GraphDatabase.driver(dburl,
                                  AuthTokens.basic(user, passw),
                                  Config.build().withMaxSessions(maxTotal).toConfig() );
}
 

import org.apache.nifi.annotation.lifecycle.OnEnabled; //导入依赖的package包/类
/**
 * @param context
 *            the configuration context
 * @throws InitializationException
 *             if unable to create a database connection
 */
@OnEnabled
public void onEnabled(final ConfigurationContext context) throws InitializationException {
}