/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */

package info.avalon566.shardingscaling.core.job;

import info.avalon566.shardingscaling.core.config.RdbmsConfiguration;
import info.avalon566.shardingscaling.core.config.SyncConfiguration;
import info.avalon566.shardingscaling.core.config.SyncType;
import info.avalon566.shardingscaling.core.job.schedule.Event;
import info.avalon566.shardingscaling.core.job.schedule.EventType;
import info.avalon566.shardingscaling.core.job.schedule.Reporter;
import info.avalon566.shardingscaling.core.job.schedule.local.LocalSyncJobExecutor;
import info.avalon566.shardingscaling.core.sync.reader.LogPosition;
import info.avalon566.shardingscaling.core.sync.reader.ReaderFactory;
import info.avalon566.shardingscaling.core.sync.util.DataSourceFactory;
import info.avalon566.shardingscaling.core.sync.util.DbMetaDataUtil;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * Database sync job.
 *
 * @author avalon566
 */
@Slf4j
public class DatabaseSyncJob {

    private final HistoryDataSyncer historyDataSyncer;

    private final RealtimeDataSyncJob realtimeDataSyncJob;

    private final Reporter reporter;

    private final SyncConfiguration syncConfiguration;

    public DatabaseSyncJob(final SyncConfiguration syncConfiguration, final Reporter reporter) {
        this.syncConfiguration = syncConfiguration;
        this.reporter = reporter;
        this.historyDataSyncer = new HistoryDataSyncer(syncConfiguration);
        this.realtimeDataSyncJob = new RealtimeDataSyncJob(syncConfiguration, reporter);

    }

    /**
     * Run.
     */
    public void run() {
        LogPosition position = realtimeDataSyncJob.preRun();
        historyDataSyncer.run();
        syncConfiguration.setPosition(position);
        SyncConfiguration realConfiguration = new SyncConfiguration(
                SyncType.Realtime, syncConfiguration.getConcurrency(),
                syncConfiguration.getReaderConfiguration(),
                syncConfiguration.getWriterConfiguration());
        realConfiguration.setPosition(position);
        Reporter realtimeReporter = new LocalSyncJobExecutor().execute(Collections.singletonList(realConfiguration));
        Event event = realtimeReporter.consumeEvent();
        if (EventType.FINISHED == event.getEventType()) {
            return;
        }
        if (EventType.EXCEPTION_EXIT == event.getEventType()) {
            System.exit(1);
        }
    }

    /**
     * History data syncer.
     *
     * @author avalon566
     */
    @Slf4j
    private static class HistoryDataSyncer {

        private final SyncConfiguration syncConfiguration;

        public HistoryDataSyncer(final SyncConfiguration syncConfiguration) {
            this.syncConfiguration = syncConfiguration;
        }

        /**
         * Run.
         */
        public final void run() {
            List<SyncConfiguration> configs = split(syncConfiguration);
            Reporter reporter = new LocalSyncJobExecutor().execute(configs);
            waitSlicesFinished(configs, reporter);
        }

        private List<SyncConfiguration> split(final SyncConfiguration syncConfiguration) {
            List<SyncConfiguration> syncConfigurations = new ArrayList<>();
            // split by table
            DataSource dataSource = DataSourceFactory.getDataSource(syncConfiguration.getReaderConfiguration().getDataSourceConfiguration());
            for (String tableName : new DbMetaDataUtil(dataSource).getTableNames()) {
                RdbmsConfiguration readerConfig = RdbmsConfiguration.clone(syncConfiguration.getReaderConfiguration());
                readerConfig.setTableName(tableName);
                // split by primary key range
                for (RdbmsConfiguration sliceConfig : ReaderFactory.newInstanceJdbcReader(readerConfig).split(syncConfiguration.getConcurrency())) {
                    syncConfigurations.add(new SyncConfiguration(SyncType.TableSlice, syncConfiguration.getConcurrency(),
                            sliceConfig, RdbmsConfiguration.clone(syncConfiguration.getWriterConfiguration())));
                }
            }
            return syncConfigurations;
        }

        private void waitSlicesFinished(final List<SyncConfiguration> syncConfigurations, final Reporter reporter) {
            int counter = 0;
            boolean hasException = false;
            while (true) {
                Event event = reporter.consumeEvent();
                if (EventType.FINISHED == event.getEventType()) {
                    counter++;
                }
                if (EventType.EXCEPTION_EXIT == event.getEventType()) {
                    hasException = true;
                    System.exit(1);
                }
                if (syncConfigurations.size() == counter) {
                    log.info("history data sync finish");
                    break;
                }
            }
        }
    }
}

@@ -1865,197 +1865,81 @@
 nal 
-HistoryDataSyncer historyDataSyncer;%0A%0A    private final RealtimeDataSyncJob realtimeDataSyncJob;%0A%0A    private final Reporter reporter;%0A%0A    private final SyncConfiguration syncConfiguration
+SyncConfiguration syncConfiguration;%0A%0A    private final Reporter reporter
 ;%0A%0A 
@@ -2124,173 +2124,8 @@
 er;%0A
-        this.historyDataSyncer = new HistoryDataSyncer(syncConfiguration);%0A        this.realtimeDataSyncJob = new RealtimeDataSyncJob(syncConfiguration, reporter);%0A%0A
     
@@ -2210,17 +2210,21 @@
 ition = 
-r
+new R
 ealtimeD
@@ -2233,16 +2233,41 @@
 aSyncJob
+(syncConfiguration, null)
 .preRun(
@@ -2277,17 +2277,21 @@
         
-h
+new H
 istoryDa
@@ -2298,16 +2298,35 @@
 taSyncer
+(syncConfiguration)
 .run();%0A

Refactor database sync job