remove async job

sharding-jdbc-transaction-parent/sharding-jdbc-transaction/src/main/java/io/shardingjdbc/transaction/api/SoftTransactionManager.java

@@ -17,17 +17,16 @@
 
 package io.shardingjdbc.transaction.api;
 
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
 import io.shardingjdbc.core.executor.threadlocal.ExecutorDataMap;
 import io.shardingjdbc.core.util.EventBusInstance;
 import io.shardingjdbc.transaction.api.config.SoftTransactionConfiguration;
 import io.shardingjdbc.transaction.bed.BEDSoftTransaction;
-import io.shardingjdbc.transaction.bed.async.NestedBestEffortsDeliveryJobFactory;
 import io.shardingjdbc.transaction.bed.sync.BestEffortsDeliveryListener;
 import io.shardingjdbc.transaction.constants.SoftTransactionType;
 import io.shardingjdbc.transaction.constants.TransactionLogDataSourceType;
 import io.shardingjdbc.transaction.tcc.TCCSoftTransaction;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
@@ -62,9 +61,6 @@ public final class SoftTransactionManager {
             Preconditions.checkNotNull(transactionConfig.getTransactionLogDataSource());
             createTable();
         }
-        if (transactionConfig.getBestEffortsDeliveryJobConfiguration().isPresent()) {
-            new NestedBestEffortsDeliveryJobFactory(transactionConfig).init();
-        }
     }
     
     private void createTable() throws SQLException {

sharding-jdbc-transaction-parent/sharding-jdbc-transaction/src/main/java/io/shardingjdbc/transaction/bed/async/AbstractBestEffortsDeliveryJobFactory.java

-/*
- * Copyright 1999-2015 dangdang.com.
- * <p>
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * </p>
- */
-
-package io.shardingjdbc.transaction.bed.async;
-
-import com.dangdang.ddframe.job.api.JobConfiguration;
-import com.dangdang.ddframe.job.api.JobScheduler;
-import io.shardingjdbc.transaction.api.config.AbstractBestEffortsDeliveryJobConfiguration;
-import io.shardingjdbc.transaction.api.config.SoftTransactionConfiguration;
-import io.shardingjdbc.transaction.storage.TransactionLogStorageFactory;
-import com.dangdang.ddframe.reg.base.CoordinatorRegistryCenter;
-import com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration;
-import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenter;
-import lombok.RequiredArgsConstructor;
-
-/**
- * Best efforts delivery B.A.S.E transaction asynchronized job abstract factory.
- * 
- * @author zhangliang
- */
-@RequiredArgsConstructor
-public abstract class AbstractBestEffortsDeliveryJobFactory<T extends AbstractBestEffortsDeliveryJobConfiguration> {
-    
-    private final SoftTransactionConfiguration transactionConfig;
-    
-    /**
-     * initialize asynchronized job.
-     */
-    public final void init() {
-        @SuppressWarnings("unchecked")
-        T bedJobConfig = (T) transactionConfig.getBestEffortsDeliveryJobConfiguration().get();
-        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(createZookeeperConfiguration(bedJobConfig));
-        regCenter.init();
-        JobScheduler jobScheduler = new JobScheduler(regCenter, createBedJobConfiguration(bedJobConfig));
-        jobScheduler.setField("transactionConfig", transactionConfig);
-        jobScheduler.setField("transactionLogStorage", TransactionLogStorageFactory.createTransactionLogStorage(transactionConfig.buildTransactionLogDataSource()));
-        jobScheduler.init();
-    }
-    
-    protected abstract ZookeeperConfiguration createZookeeperConfiguration(T config);
-    
-    private JobConfiguration createBedJobConfiguration(final T bedJobConfig) {
-        JobConfiguration result = new JobConfiguration(bedJobConfig.getJobName(), NestedBestEffortsDeliveryJob.class, 1, bedJobConfig.getCron());
-        result.setFetchDataCount(bedJobConfig.getTransactionLogFetchDataCount());
-        return result;
-    }
-}

sharding-jdbc-transaction-parent/sharding-jdbc-transaction/src/main/java/io/shardingjdbc/transaction/bed/async/NestedBestEffortsDeliveryJob.java

-/*
- * Copyright 1999-2015 dangdang.com.
- * <p>
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * </p>
- */
-
-package io.shardingjdbc.transaction.bed.async;
-
-import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
-import com.dangdang.ddframe.job.plugin.job.type.dataflow.AbstractIndividualThroughputDataFlowElasticJob;
-import io.shardingjdbc.core.exception.ShardingJdbcException;
-import io.shardingjdbc.transaction.api.config.SoftTransactionConfiguration;
-import io.shardingjdbc.transaction.storage.TransactionLog;
-import io.shardingjdbc.transaction.storage.TransactionLogStorage;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-
-import java.sql.SQLException;
-import java.util.List;
-
-/**
- * Best efforts delivery B.A.S.E transaction asynchronized job.
- * 
- * @author zhangliang
- * @author caohao
- */
-@Slf4j
-public class NestedBestEffortsDeliveryJob extends AbstractIndividualThroughputDataFlowElasticJob<TransactionLog> {
-    
-    @Setter
-    private SoftTransactionConfiguration transactionConfig;
-    
-    @Setter
-    private TransactionLogStorage transactionLogStorage;
-    
-    @Override
-    public List<TransactionLog> fetchData(final JobExecutionMultipleShardingContext context) {
-        return transactionLogStorage.findEligibleTransactionLogs(context.getFetchDataCount(), 
-            transactionConfig.getBestEffortsDeliveryJobConfiguration().get().getAsyncMaxDeliveryTryTimes(), 
-            transactionConfig.getBestEffortsDeliveryJobConfiguration().get().getAsyncMaxDeliveryTryDelayMillis());
-    }
-    
-    @Override
-    public boolean processData(final JobExecutionMultipleShardingContext context, final TransactionLog data) {
-        try {
-            return transactionLogStorage.processData(
-                    transactionConfig.getTargetConnection(data.getDataSource()), data, transactionConfig.getBestEffortsDeliveryJobConfiguration().get().getAsyncMaxDeliveryTryTimes());
-        } catch (final SQLException ex) {
-            throw new ShardingJdbcException(ex);
-        }
-    }
-    
-    @Override
-    public boolean isStreamingProcess() {
-        return false;
-    }
-}

sharding-jdbc-transaction-parent/sharding-jdbc-transaction/src/main/java/io/shardingjdbc/transaction/bed/async/NestedBestEffortsDeliveryJobFactory.java

-/*
- * Copyright 1999-2015 dangdang.com.
- * <p>
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * </p>
- */
-
-package io.shardingjdbc.transaction.bed.async;
-
-import io.shardingjdbc.transaction.api.config.NestedBestEffortsDeliveryJobConfiguration;
-import io.shardingjdbc.transaction.api.config.SoftTransactionConfiguration;
-import com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration;
-
-/**
- * Best efforts delivery B.A.S.E transaction asynchronized job factory.
- * 
- * @author zhangliang
- */
-public final class NestedBestEffortsDeliveryJobFactory extends AbstractBestEffortsDeliveryJobFactory<NestedBestEffortsDeliveryJobConfiguration> {
-    
-    public NestedBestEffortsDeliveryJobFactory(final SoftTransactionConfiguration transactionConfig) {
-        super(transactionConfig);
-    }
-    
-    @Override
-    protected ZookeeperConfiguration createZookeeperConfiguration(final NestedBestEffortsDeliveryJobConfiguration config) {
-        ZookeeperConfiguration result = new ZookeeperConfiguration(String.format("localhost:%s", config.getZookeeperPort()), 
-                config.getJobNamespace(), config.getZookeeperBaseSleepTimeMilliseconds(), config.getZookeeperMaxSleepTimeMilliseconds(), config.getZookeeperMaxRetries());
-        result.setNestedPort(config.getZookeeperPort());
-        result.setNestedDataDir(config.getZookeeperDataDir());
-        return result;
-    }
-}