tencent-baidu-tracking/src/main/java/com/adx/tencent/oppo/service/OppoBackgroundTasks.java

package com.adx.tencent.oppo.service;

import com.adx.tencent.config.AppProperties;
import com.adx.tencent.leader.LeaderElection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

@Component
public class OppoBackgroundTasks implements ApplicationRunner {
    private static final Logger log = LoggerFactory.getLogger(OppoBackgroundTasks.class);

    @Autowired private AppProperties props;
    @Autowired(required = false) private OppoColdWorker worker;
    @Autowired(required = false) private OppoRetryService retryService;
    @Autowired(required = false) private OppoTagEventSyncService tagSync;
    @Autowired(required = false) @Nullable private LeaderElection leaderElection;

    private final AtomicBoolean stopped = new AtomicBoolean();
    private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
        Thread t = new Thread(r);
        t.setDaemon(true);
        return t;
    });

    public void run(ApplicationArguments args) {
        if (worker != null) {
            executor.submit(() -> {
                while (!stopped.get()) {
                    try {
                        if (worker.runOnce() == 0) {
                            sleep(200);
                        }
                    } catch (Exception e) {
                        log.error("oppo cold worker: {}", e.getMessage(), e);
                        sleep(1000);
                    }
                }
            });
        }

        if (retryService != null && props.isOppoCallbackRetryEnabled()) {
            if (props.isSkipLeaderElection() || leaderElection == null) {
                executor.submit(() -> runRetry(() -> stopped.get()));
            } else {
                executor.submit(() -> leaderElection.run(
                        "adx:lock:oppo:callback-retry",
                        props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
                        stopped::get,
                        this::runRetry,
                        e -> log.error("oppo callback retry leader: {}", e.getMessage(), e)
                ));
            }
        }

        if (tagSync != null && props.isOppoTagEventSyncEnabled()) {
            if (props.isSkipLeaderElection() || leaderElection == null) {
                executor.submit(() -> runTagEventSync(() -> stopped.get()));
            } else {
                executor.submit(() -> leaderElection.run(
                        "adx:lock:oppo:tag-event-sync",
                        props.getTaskLockTtl(), props.getTaskLockRenewInterval(), props.getTaskLockRetryInterval(),
                        stopped::get,
                        this::runTagEventSync,
                        e -> log.error("oppo tag event sync leader: {}", e.getMessage(), e)
                ));
            }
        }
    }

    private void runRetry(LeaderElection.StopSignal stopSignal) {
        long intervalMs = props.getOppoCallbackRetryInterval().toMillis();
        while (!stopSignal.isStopped()) {
            try {
                retryService.retry(props.getOppoCallbackRetryLimit());
            } catch (Exception e) {
                log.error("oppo callback retry: {}", e.getMessage(), e);
            }
            sleepResponsive(intervalMs, stopSignal);
        }
    }

    private void runTagEventSync(LeaderElection.StopSignal stopSignal) {
        long intervalMs = props.getOppoTagEventSyncInterval().toMillis();
        while (!stopSignal.isStopped()) {
            try {
                tagSync.syncOnce();
            } catch (Exception e) {
                log.error("oppo tag event sync: {}", e.getMessage(), e);
            }
            sleepResponsive(intervalMs, stopSignal);
        }
    }

    private void sleepResponsive(long millis, LeaderElection.StopSignal stopSignal) {
        long remaining = millis;
        while (remaining > 0 && !stopSignal.isStopped()) {
            long chunk = Math.min(remaining, 1000);
            sleep(chunk);
            remaining -= chunk;
        }
    }

    private void sleep(long millis) {
        try {
            Thread.sleep(Math.max(1, millis));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            stopped.set(true);
        }
    }
}