Skip to content

Commit

Permalink
Merge pull request #2310 from terrymanu/dev
Browse files Browse the repository at this point in the history
Use ShardingSphereServiceLoader to instead of JDK ServiceLoader
  • Loading branch information
sandynz authored Oct 25, 2023
2 parents caef949 + 175a2ff commit f9825c7
Show file tree
Hide file tree
Showing 30 changed files with 95 additions and 313 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,7 @@
import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.executor.item.impl.ClassedJobItemExecutor;
import org.apache.shardingsphere.elasticjob.infra.exception.JobConfigurationException;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.ServiceLoader;
import org.apache.shardingsphere.infra.spi.ShardingSphereServiceLoader;

/**
* Job item executor factory.
Expand All @@ -35,12 +31,6 @@
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class JobItemExecutorFactory {

private static final Map<Class, ClassedJobItemExecutor> CLASSED_EXECUTORS = new HashMap<>();

static {
ServiceLoader.load(ClassedJobItemExecutor.class).forEach(each -> CLASSED_EXECUTORS.put(each.getElasticJobClass(), each));
}

/**
* Get executor.
*
Expand All @@ -49,9 +39,9 @@ public final class JobItemExecutorFactory {
*/
@SuppressWarnings("unchecked")
public static JobItemExecutor getExecutor(final Class<? extends ElasticJob> elasticJobClass) {
for (Entry<Class, ClassedJobItemExecutor> entry : CLASSED_EXECUTORS.entrySet()) {
if (entry.getKey().isAssignableFrom(elasticJobClass)) {
return entry.getValue();
for (ClassedJobItemExecutor each : ShardingSphereServiceLoader.getServiceInstances(ClassedJobItemExecutor.class)) {
if (each.getElasticJobClass().isAssignableFrom(elasticJobClass)) {
return each;
}
}
throw new JobConfigurationException("Can not find executor for elastic job class `%s`", elasticJobClass.getName());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,14 @@

import org.apache.shardingsphere.elasticjob.api.ElasticJob;
import org.apache.shardingsphere.elasticjob.executor.item.JobItemExecutor;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;

/**
* Classed job item executor.
*
* @param <T> type of ElasticJob
*/
@SingletonSPI
public interface ClassedJobItemExecutor<T extends ElasticJob> extends JobItemExecutor<T> {

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,8 @@
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
import org.apache.shardingsphere.elasticjob.tracing.event.JobEvent;
import org.apache.shardingsphere.elasticjob.tracing.exception.TracingConfigurationException;
import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListenerFactory;
import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListenerConfiguration;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
Expand Down Expand Up @@ -64,9 +65,14 @@ private static ExecutorService createExecutorService(final int threadSize) {
return MoreExecutors.listeningDecorator(MoreExecutors.getExitingExecutorService(threadPoolExecutor));
}

@SuppressWarnings("unchecked")
private void register(final TracingConfiguration<?> tracingConfig) {
try {
eventBus.register(TracingListenerFactory.getListener(tracingConfig));
if (null == tracingConfig.getTracingStorageConfiguration()) {
throw new TracingConfigurationException(String.format("Can not find executor service handler type '%s'.", tracingConfig.getType()));
}
eventBus.register(
TypedSPILoader.getService(TracingListenerConfiguration.class, tracingConfig.getType()).createTracingListener(tracingConfig.getTracingStorageConfiguration().getStorage()));
isRegistered = true;
} catch (final TracingConfigurationException ex) {
log.error("Elastic job: create tracing listener failure, error is: ", ex);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,16 @@
package org.apache.shardingsphere.elasticjob.tracing.listener;

import org.apache.shardingsphere.elasticjob.tracing.exception.TracingConfigurationException;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;

/**
* Tracing listener configuration.
*
* @param <T> type of tracing storage
*/
public interface TracingListenerConfiguration<T> {
@SingletonSPI
public interface TracingListenerConfiguration<T> extends TypedSPI {

/**
* Create tracing listener.
Expand All @@ -35,10 +38,6 @@ public interface TracingListenerConfiguration<T> {
*/
TracingListener createTracingListener(T storage) throws TracingConfigurationException;

/**
* Get tracing type.
*
* @return tracing type
*/
@Override
String getType();
}

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@
package org.apache.shardingsphere.elasticjob.tracing.storage;

import org.apache.shardingsphere.elasticjob.tracing.api.TracingStorageConfiguration;
import org.apache.shardingsphere.infra.spi.annotation.SingletonSPI;

/**
* Tracing storage converter.
*
* @param <T> storage type
*/
@SingletonSPI
public interface TracingStorageConverter<T> {

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,33 +19,26 @@

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.shardingsphere.infra.spi.ShardingSphereServiceLoader;

import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.ServiceLoader;

/**
* Factory for {@link TracingStorageConverter}.
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class TracingStorageConverterFactory {

private static final List<TracingStorageConverter<?>> CONVERTERS = new LinkedList<>();

static {
ServiceLoader.load(TracingStorageConverter.class).forEach(CONVERTERS::add);
}

/**
* Find {@link TracingStorageConverter} for specific storage type.
*
* @param storageType storage type
* @param <T> storage type
* @param <T> storage type
* @return instance of {@link TracingStorageConverter}
*/
@SuppressWarnings("unchecked")
public static <T> Optional<TracingStorageConverter<T>> findConverter(final Class<T> storageType) {
return CONVERTERS.stream().filter(each -> each.storageType().isAssignableFrom(storageType)).map(each -> (TracingStorageConverter<T>) each).findFirst();
return ShardingSphereServiceLoader.getServiceInstances(TracingStorageConverter.class).stream()
.filter(each -> each.storageType().isAssignableFrom(storageType)).map(each -> (TracingStorageConverter<T>) each).findFirst();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,9 @@
package org.apache.shardingsphere.elasticjob.tracing.yaml;

import org.apache.shardingsphere.elasticjob.infra.yaml.config.YamlConfigurationConverter;
import org.apache.shardingsphere.elasticjob.infra.yaml.config.YamlConfigurationConverterFactory;
import org.apache.shardingsphere.elasticjob.infra.yaml.exception.YamlConfigurationConverterNotFoundException;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingConfiguration;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingStorageConfiguration;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;

/**
* Converter to convert {@link TracingConfiguration} to {@link YamlTracingConfiguration}.
Expand All @@ -32,21 +31,19 @@
public final class YamlTracingConfigurationConverter<T> implements YamlConfigurationConverter<TracingConfiguration<T>, YamlTracingConfiguration<T>> {

@Override
public YamlTracingConfiguration<T> convertToYamlConfiguration(final TracingConfiguration<T> tracingConfiguration) {
public YamlTracingConfiguration<T> convertToYamlConfiguration(final TracingConfiguration<T> tracingConfig) {
YamlTracingConfiguration<T> result = new YamlTracingConfiguration<>();
result.setType(tracingConfiguration.getType());
result.setTracingStorageConfiguration(convertTracingStorageConfiguration(tracingConfiguration.getTracingStorageConfiguration()));
result.setType(tracingConfig.getType());
result.setTracingStorageConfiguration(convertTracingStorageConfiguration(tracingConfig.getTracingStorageConfiguration()));
return result;
}

private YamlTracingStorageConfiguration<T> convertTracingStorageConfiguration(final TracingStorageConfiguration<T> tracingStorageConfiguration) {
return YamlConfigurationConverterFactory
.<TracingStorageConfiguration<T>, YamlTracingStorageConfiguration<T>>findConverter((Class<TracingStorageConfiguration<T>>) tracingStorageConfiguration.getClass())
.orElseThrow(() -> new YamlConfigurationConverterNotFoundException(tracingStorageConfiguration.getClass())).convertToYamlConfiguration(tracingStorageConfiguration);
private YamlTracingStorageConfiguration<T> convertTracingStorageConfiguration(final TracingStorageConfiguration<T> tracingStorageConfig) {
return (YamlTracingStorageConfiguration) TypedSPILoader.getService(YamlConfigurationConverter.class, tracingStorageConfig.getClass()).convertToYamlConfiguration(tracingStorageConfig);
}

@Override
public Class configurationType() {
public Class getType() {
return TracingConfiguration.class;
}
}

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ public YamlTracingStorageConfiguration<JobEventCaller> convertToYamlConfiguratio
}

@Override
public Class configurationType() {
public Class getType() {
return JobEventCallerConfiguration.class;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.apache.shardingsphere.elasticjob.tracing.api.TracingStorageConfiguration;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;

import javax.sql.DataSource;
import java.lang.reflect.Method;
Expand All @@ -35,7 +36,6 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.ServiceLoader;

/**
* Data source configuration.
Expand Down Expand Up @@ -121,20 +121,10 @@ public DataSource createDataSource() {
setterMethod.get().invoke(result, entry.getValue());
}
}
Optional<JDBCParameterDecorator> decorator = findJDBCParameterDecorator(result);
Optional<JDBCParameterDecorator> decorator = TypedSPILoader.findService(JDBCParameterDecorator.class, result.getClass());
return decorator.isPresent() ? decorator.get().decorate(result) : result;
}

@SuppressWarnings("rawtypes")
private Optional<JDBCParameterDecorator> findJDBCParameterDecorator(final DataSource dataSource) {
for (JDBCParameterDecorator each : ServiceLoader.load(JDBCParameterDecorator.class)) {
if (each.getType() == dataSource.getClass()) {
return Optional.of(each);
}
}
return Optional.empty();
}

private Optional<Method> findSetterMethod(final Method[] methods, final String property) {
String setterMethodName = Joiner.on("").join(SETTER_PREFIX, CaseFormat.LOWER_CAMEL.to(CaseFormat.UPPER_CAMEL, property));
for (Method each : methods) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,16 @@

package org.apache.shardingsphere.elasticjob.tracing.rdb.datasource;

import org.apache.shardingsphere.infra.spi.type.typed.TypedSPI;

import javax.sql.DataSource;

/**
* JDBC parameter decorator.
*
* @param <T> type of data source
*/
public interface JDBCParameterDecorator<T extends DataSource> {
public interface JDBCParameterDecorator<T extends DataSource> extends TypedSPI {

/**
* Decorate data source.
Expand All @@ -34,10 +36,6 @@ public interface JDBCParameterDecorator<T extends DataSource> {
*/
T decorate(T dataSource);

/**
* Get data source type.
*
* @return data source type
*/
@Override
Class<T> getType();
}
Loading

0 comments on commit f9825c7

Please sign in to comment.