集成平台下连接器设计规范: 调研-实践-思考

protected abstract Endpoint<E> createEndpoint(String uri, String remaining, Map parameters)

public BlockingQueue<Exchange> createQueue(String uri, Map parameters) {
        int size = 1000;
        Object value = parameters.remove("size");
        if (value != null) {
            Integer i = convertTo(Integer.class, value);
            if (i != null) {
                size = i;
            }
        }
        return new LinkedBlockingQueue<Exchange>(size);
    }

/**
 * A Camel Java DSL Router
 */
public class MyRouteBuilder extends RouteBuilder {
    /**
     * Let's configure the Camel routing rules using Java code...
     */
    public void configure() {
        // here is a sample which processes the input files
        // (leaving them in place - see the 'noop' flag)
        // then performs content based routing on the message using XPath
        from("file:src/data?noop=true")
            .choice()
                .when(xpath("/person/city = 'London'"))
                    .to("file:target/messages/uk")
                .otherwise()
                    .to("file:target/messages/others");
    }
}

mvn org.mule.extensions:mule-extensions-archetype-maven-plugin:generate

protected void postProcessCamelContext(CamelContext camelContext) throws Exception {
        // gathers the properties (key=value) that was used as property placeholders during bootstrap
        final OrderedLocationProperties propertyPlaceholders = new OrderedLocationProperties();
        // use the main autowired lifecycle strategy instead of the default
        camelContext.getLifecycleStrategies().removeIf(s -> s instanceof AutowiredLifecycleStrategy);
        camelContext.addLifecycleStrategy(new MainAutowiredLifecycleStrategy(camelContext));
        // setup properties
        configurePropertiesService(camelContext);
        // register listener on properties component so we can capture them
        PropertiesComponent pc = camelContext.getPropertiesComponent();
        pc.addPropertiesLookupListener(new PropertyPlaceholderListener(propertyPlaceholders));
        // setup startup recorder before building context
        configureStartupRecorder(camelContext);
        // setup package scan
        configurePackageScan(camelContext);
        // configure to use our main routes loader
        configureRoutesLoader(camelContext);
        // ensure camel context is build
        camelContext.build();
        for (MainListener listener : listeners) {
            listener.beforeInitialize(this);
        }
        // allow doing custom configuration before camel is started
        for (MainListener listener : listeners) {
            listener.beforeConfigure(this);
        }
        // we want to capture startup events for import tasks during main bootstrap
        StartupStepRecorder recorder = camelContext.adapt(ExtendedCamelContext.class).getStartupStepRecorder();
        StartupStep step;
        if (standalone) {
            step = recorder.beginStep(BaseMainSupport.class, "autoconfigure", "Auto Configure");
            autoconfigure(camelContext);
            recorder.endStep(step);
        }
        if (mainConfigurationProperties.isEagerClassloading()) {
            step = recorder.beginStep(BaseMainSupport.class, "classloading", "Eager Classloading");
            EagerClassloadedHelper.eagerLoadClasses();
            recorder.endStep(step);
        }
        configureLifecycle(camelContext);
        if (standalone) {
            step = recorder.beginStep(BaseMainSupport.class, "configureRoutes", "Collect Routes");
            configureRoutes(camelContext);
            recorder.endStep(step);
        }
        // allow doing custom configuration before camel is started
        for (MainListener listener : listeners) {
            listener.afterConfigure(this);
            listener.configure(camelContext);
        }
        // we want to log the property placeholder summary after routes has been started,
        // but before camel context logs that it has been started, so we need to use an event listener
        if (standalone && mainConfigurationProperties.isAutoConfigurationLogSummary()) {
            camelContext.getManagementStrategy().addEventNotifier(new EventNotifierSupport() {
                @Override
                public boolean isEnabled(CamelEvent event) {
                    return event instanceof CamelContextRoutesStartedEvent;
                }
                @Override
                public void notify(CamelEvent event) throws Exception {
                    // log summary of configurations
                    if (!propertyPlaceholders.isEmpty()) {
                        LOG.info("Property-placeholders summary");
                        for (var entry : propertyPlaceholders.entrySet()) {
                            String k = entry.getKey().toString();
                            Object v = entry.getValue();
                            String loc = locationSummary(propertyPlaceholders, k);
                            if (SensitiveUtils.containsSensitive(k)) {
                                LOG.info("    {} {}=xxxxxx", loc, k);
                            } else {
                                LOG.info("    {} {}={}", loc, k, v);
                            }
                        }
                    }
                }
            });
        }
    }

protected void configureRoutes(CamelContext camelContext) throws Exception {
        // then configure and add the routes
        RoutesConfigurer configurer = new RoutesConfigurer();
        if (mainConfigurationProperties.isRoutesCollectorEnabled()) {
            configurer.setRoutesCollector(routesCollector);
        }
        configurer.setBeanPostProcessor(camelContext.adapt(ExtendedCamelContext.class).getBeanPostProcessor());
        configurer.setRoutesBuilders(mainConfigurationProperties.getRoutesBuilders());
        configurer.setRoutesBuilderClasses(mainConfigurationProperties.getRoutesBuilderClasses());
        if (mainConfigurationProperties.isBasePackageScanEnabled()) {
            // only set the base package if enabled
            configurer.setBasePackageScan(mainConfigurationProperties.getBasePackageScan());
        }
        configurer.setJavaRoutesExcludePattern(mainConfigurationProperties.getJavaRoutesExcludePattern());
        configurer.setJavaRoutesIncludePattern(mainConfigurationProperties.getJavaRoutesIncludePattern());
        configurer.setRoutesExcludePattern(mainConfigurationProperties.getRoutesExcludePattern());
        configurer.setRoutesIncludePattern(mainConfigurationProperties.getRoutesIncludePattern());
        configurer.configureRoutes(camelContext);
    }

public void configureRoutes(CamelContext camelContext) throws Exception {
        final List<RoutesBuilder> routes = new ArrayList<>();
        if (getRoutesBuilders() != null) {
            routes.addAll(getRoutesBuilders());
        }
        if (getRoutesBuilderClasses() != null) {
            String[] routeClasses = getRoutesBuilderClasses().split(",");
            for (String routeClass : routeClasses) {
                Class<RoutesBuilder> routeClazz = camelContext.getClassResolver().resolveClass(routeClass, RoutesBuilder.class);
                if (routeClazz == null) {
                    LOG.warn("Unable to resolve class: {}", routeClass);
                    continue;
                }
                // lets use Camel's injector so the class has some support for dependency injection
                RoutesBuilder builder = camelContext.getInjector().newInstance(routeClazz);
                routes.add(builder);
            }
        }
        if (getBasePackageScan() != null) {
            String[] pkgs = getBasePackageScan().split(",");
            Set<Class<?>> set = camelContext.adapt(ExtendedCamelContext.class)
                    .getPackageScanClassResolver()
                    .findImplementations(RoutesBuilder.class, pkgs);
            for (Class<?> routeClazz : set) {
                Object builder = camelContext.getInjector().newInstance(routeClazz);
                if (builder instanceof RoutesBuilder) {
                    routes.add((RoutesBuilder) builder);
                } else {
                    LOG.warn("Class {} is not a RouteBuilder class", routeClazz);
                }
            }
        }
        if (getRoutesCollector() != null) {
            try {
                LOG.debug("RoutesCollectorEnabled: {}", getRoutesCollector());
                // add discovered routes from registry
                Collection<RoutesBuilder> routesFromRegistry = getRoutesCollector().collectRoutesFromRegistry(
                        camelContext,
                        getJavaRoutesExcludePattern(),
                        getJavaRoutesIncludePattern());
                routes.addAll(routesFromRegistry);
                if (LOG.isDebugEnabled() && !routesFromRegistry.isEmpty()) {
                    LOG.debug("Discovered {} additional RoutesBuilder from registry: {}", routesFromRegistry.size(),
                            getRoutesIncludePattern());
                }
                // add discovered routes from directories
                StopWatch watch = new StopWatch();
                Collection<RoutesBuilder> routesFromDirectory = getRoutesCollector().collectRoutesFromDirectory(
                        camelContext,
                        getRoutesExcludePattern(),
                        getRoutesIncludePattern());
                routes.addAll(routesFromDirectory);
                if (LOG.isDebugEnabled() && !routesFromDirectory.isEmpty()) {
                    LOG.debug("Loaded {} additional RoutesBuilder from: {} (took {})", routesFromDirectory.size(),
                            getRoutesIncludePattern(), TimeUtils.printDuration(watch.taken(), true));
                }
            } catch (Exception e) {
                throw RuntimeCamelException.wrapRuntimeException(e);
            }
        }
        if (getBeanPostProcessor() != null) {
            // lets use Camel's bean post processor on any existing route builder classes
            // so the instance has some support for dependency injection
            for (RoutesBuilder routeBuilder : routes) {
                getBeanPostProcessor().postProcessBeforeInitialization(routeBuilder, routeBuilder.getClass().getName());
                getBeanPostProcessor().postProcessAfterInitialization(routeBuilder, routeBuilder.getClass().getName());
            }
        }
        // add the discovered routes
        addDiscoveredRoutes(camelContext, routes);
        // then discover and add templates
        Set<ConfigureRouteTemplates> set = camelContext.getRegistry().findByType(ConfigureRouteTemplates.class);
        for (ConfigureRouteTemplates crt : set) {
            LOG.debug("Configuring route templates via: {}", crt);
            crt.configure(camelContext);
        }
    }

public void start(boolean registerShutdownHook) throws MuleException {
    if (registerShutdownHook) {
      registerShutdownHook();
    }
    try {
      startIfNeeded(artifactResourcesRegistry.getContainerProfilingService());
      doResourceInitialization();
      createExecutionMuleFolder();
      serviceManager.start();
      coreExtensionManager.setDeploymentService(deploymentService);
      coreExtensionManager.setRepositoryService(repositoryService);
      coreExtensionManager.setArtifactClassLoaderManager(artifactResourcesRegistry.getArtifactClassLoaderManager());
      coreExtensionManager.setToolingService(toolingService);
      coreExtensionManager.setServiceRepository(serviceManager);
      coreExtensionManager.setTroubleshootingService(troubleshootingService);
      validateLicense();
      showSplashScreen();
      coreExtensionManager.initialise();
      coreExtensionManager.start();
      toolingService.initialise();
      extensionModelLoaderManager.start();
      deploymentService.start();
    } catch (Throwable e) {
      shutdown(e);
    }
  }

/**
   * Lists all the files in the {@code directoryPath} which match the given {@code matcher}.
   * <p>
   * If the listing encounters a directory, the output list will include its contents depending on the value of the
   * {@code recursive} parameter.
   * <p>
   *
   * @param config        the config that is parameterizing this operation
   * @param directoryPath the path to the directory to be listed
   * @param recursive     whether to include the contents of sub-directories. Defaults to false.
   * @param matcher     a matcher used to filter the output list
   * @param timeBetweenSizeCheck wait time between size checks to determine if a file is ready to be read.
   * @param timeBetweenSizeCheckUnit time unit to be used in the wait time between size checks.
   * @return a {@link List} of {@link Message messages} each one containing each file's content in the payload and metadata in the attributes
   * @throws IllegalArgumentException if {@code directoryPath} points to a file which doesn't exist or is not a directory
   */
  @Summary("List all the files from given directory")
  @MediaType(value = ANY, strict = false)
  @Throws(FileListErrorTypeProvider.class)
  public PagingProvider<LocalFileSystem, Result<Object, LocalFileAttributes>> list(@Config FileConnectorConfig config,
                                                                                   @Path(type = DIRECTORY,
                                                                                       location = EXTERNAL) String directoryPath,
                                                                                   @Optional(
                                                                                       defaultValue = "false") boolean recursive,
                                                                                   @Optional @DisplayName("File Matching Rules") @Summary("Matcher to filter the listed files") LocalFileMatcher matcher,
                                                                                   @ConfigOverride @Placement(
                                                                                       tab = ADVANCED_TAB) Long timeBetweenSizeCheck,
                                                                                   @ConfigOverride @Placement(
                                                                                       tab = ADVANCED_TAB) TimeUnit timeBetweenSizeCheckUnit,
                                                                                   StreamingHelper streamingHelper,
                                                                                   @Optional @Placement(
                                                                                       tab = ADVANCED_TAB) @Summary("Limit and sort the number of files returned") LocalSubsetList subset) {
    PagingProvider result =
        doPagedList(config, directoryPath, recursive, matcher,
                    config.getTimeBetweenSizeCheckInMillis(timeBetweenSizeCheck, timeBetweenSizeCheckUnit).orElse(null),
                    streamingHelper, subset);
    return result;
  }

@Connector(name = "jdbc", title = "JDBC连接器")
public interface JdbcConnector {
    /**
     * 遍历查询全表, 不使用query中sql语句
     *
     * @param config
     * @param query  支持额外参数(AdvancedOptions):
     *               分页查询参数:
     *               int pageSize - default=10
     *               String waterColumn
     *               sql修饰参数:
     *               String selectColumns
     *               String whereCondition
     * @return
     */
    @Operation(name = "iterate", title = "遍历查询(返回map)")
    Iterator<Map<String, Object>> iterate(@Config JdbcConfig config, @Param Query query);
}
    public <T> Iterator<T> iterate(Query<T> query, JdbcConfig config) {
        NamedParameterJdbcTemplate jdbcTemplate = jdbcTemplate(config);
        String table = getTableNameFromAnnotationAndQuery(query);
        PageIteratorAsClass<T> pageIteratorAsClass = new PageIteratorAsClass<>(jdbcTemplate, query);
        return pageIteratorAsClass;
    }
}

public class PageIteratorAsClass<T> implements Iterator<T> {
    @Override
    public boolean hasNext() {
        if (iterator == null || iterator.hasNext() == false) poll();
        return iterator.hasNext();
    }
    @Override
    public T next() {
        T t = iterator.next();
        if (t != null) {
            updateDataIndex(t);
        }
        return t;
    }
    private void updateDataIndex(T data) {
        dataMark!=null?updateIndexWithDataMark():updateIndexWithOutDataMark();
    }
    /*
     * current, limit only get supported by mysql
     * so this connector only works for rds-mysql now
     */
    private void poll() {
        String sql = generateNextSQL();
        LOG.info("Execute " + sql);
        iterator = jdbcTemplate.query(sql, connectorRowMapper).stream().iterator();
    }
}

public void init() {
        jdbcConnector = new JdbcConnectorImpl();
        jdbcConfig = new JdbcConfig();
        setJdbcConfig(jdbcConfig);
    }
    public void iterate() {
        Query<MidwayData> query = new Query<>();
        query.setClazz(MidwayData.class); //只需在query中注明DO类, 即可实现全表查询
        // 更多设置可以在query中进行定义, 本方法反射方式读取mybatis注解属性构造sql语句
        Iterator<MidwayData> iterator = jdbcConnector.iterate(query, jdbcConfig);
        while (iterator.hasNext()) {
            System.out.println(iterator.next().toString());
        }
    }

public void test() throws Exception {
        FlowGeneratorImpl generator = new FlowGeneratorImpl();
        generator.setNodeMetaProvider(new NodeMetaProviderImpl());
        FlowModel flowModel = new FlowModel();
        NodeModel nodeModel = new NodeModel();
        setNodeModelProperty(nodeModel);
        NodeModel nodeModel2 = new NodeModel();
        setNodeModelProperty(nodeModel2);
        flowModel.getNodes().add(nodeModel).add(nodeModel2);
        String code = generator.generate(flowModel);
        System.out.println(code);
    }

/** codegen by engine @Date Wed Sep 28 17:47:13 CST 2022 */
public class nullFlow implements Flow {
    @Resource 
    private HttpConnector httpConnector;
    @Override
    public void execute(FlowContext context) {
        /* step 1 invoke httpTest  */
        httpTest(context);
        /* step 2 invoke httpTest2  */
        httpTest2(context);
    }
    private void httpTest(FlowContext context) {
        HttpConfig config = context.getConfig("httpConfigxxx", HttpConfig.class);
        HttpRequest request = new HttpRequest();
        request.setMethod("GET");
        HttpResponse response = httpConnector.execute(config, request);
        context.putResult("httpTest", response);
    }
    private void httpTest2(FlowContext context) {
        HttpConfig config = context.getConfig("httpConfigxxx2", HttpConfig.class);
        HttpRequest request = new HttpRequest();
        request.setMethod("POST");
        request.setBody(context.getValue("#input.body", String.class));
        request.setContentType("application/json");
        HttpResponse response = httpConnector.execute(config, request);
        context.putResult("httpTest2", response);
    }
}