<commit_before>package com.ctrip.xpipe.redis.console.healthcheck.delay;

import com.ctrip.xpipe.api.foundation.FoundationService;
import com.ctrip.xpipe.redis.console.health.RedisSession;
import com.ctrip.xpipe.redis.console.healthcheck.AbstractHealthCheckAction;
import com.ctrip.xpipe.redis.console.healthcheck.RedisHealthCheckInstance;
import com.ctrip.xpipe.redis.console.healthcheck.action.HealthStatus;
import com.ctrip.xpipe.redis.console.healthcheck.ping.PingService;
import com.ctrip.xpipe.utils.DateTimeUtils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author chen.zhu
 * <p>
 * Sep 06, 2018
 */
public class DelayAction extends AbstractHealthCheckAction<DelayActionContext> {

    public static final String CHECK_CHANNEL = "xpipe-health-check-" + FoundationService.DEFAULT.getLocalIp();

    private static final DelayActionContext INIT_CONTEXT = new DelayActionContext(null, HealthStatus.UNSET_TIME);

    public static final long SAMPLE_LOST_AND_NO_PONG = -99999L * 1000 * 1000;

    public static final long SAMPLE_LOST_BUT_PONG = 99999L * 1000 * 1000;

    private SubscribeCallback callback = new SubscribeCallback();

    private AtomicReference<DelayActionContext> context = new AtomicReference<>(INIT_CONTEXT);

    private PingService pingService;

    private final long expireInterval;

    public DelayAction(ScheduledExecutorService scheduled, RedisHealthCheckInstance instance,
                       ExecutorService executors, PingService pingService) {
        super(scheduled, instance, executors);
        this.pingService = pingService;
        expireInterval = instance.getHealthCheckConfig().getHealthyDelayMilli() + DELTA * 2;
    }

    @Override
    protected void doScheduledTask() {
        reportDelay();
        RedisSession session = instance.getRedisSession();
        session.subscribeIfAbsent(CHECK_CHANNEL, callback);
        if(instance.getRedisInstanceInfo().isMaster()) {
            session.publish(CHECK_CHANNEL, Long.toHexString(System.nanoTime()));
        }
    }

    private void reportDelay() {
        if(INIT_CONTEXT.equals(context.get())) {
            return;
        }
        if(isExpired()) {
            logger.warn("[expire][{}] last update time: {}", instance.getRedisInstanceInfo().getHostPort(),
                    DateTimeUtils.timeAsString(TimeUnit.NANOSECONDS.toMillis(context.get().getRecvTimeNano())));

            long result = SAMPLE_LOST_AND_NO_PONG;
            if(pingService.isRedisAlive(instance.getRedisInstanceInfo().getHostPort())) {
                result = SAMPLE_LOST_BUT_PONG;
            }
            notifyListeners(new DelayActionContext(instance, result));
        } else {
            notifyListeners(context.get());
        }
    }

    private class SubscribeCallback implements RedisSession.SubscribeCallback {

        @Override
        public void message(String channel, String message) {
            onMessage(message);
        }

        @Override
        public void fail(Throwable e) {
            //ignore sub fail
        }
    }

    private void onMessage(String message) {
        long currentTime = System.nanoTime();
        long lastDelayPubTimeNano = Long.parseLong(message, 16);
        this.context.set(new DelayActionContext(instance, currentTime - lastDelayPubTimeNano));
    }

    private boolean isExpired() {
        long recvMilli = TimeUnit.NANOSECONDS.toMillis(context.get().getRecvTimeNano());
        return System.currentTimeMillis() - recvMilli >= expireInterval;
    }

}<commit_msg>acurate expire time<commit_after>package com.ctrip.xpipe.redis.console.healthcheck.delay;

import com.ctrip.xpipe.api.foundation.FoundationService;
import com.ctrip.xpipe.redis.console.health.RedisSession;
import com.ctrip.xpipe.redis.console.healthcheck.AbstractHealthCheckAction;
import com.ctrip.xpipe.redis.console.healthcheck.RedisHealthCheckInstance;
import com.ctrip.xpipe.redis.console.healthcheck.action.HealthStatus;
import com.ctrip.xpipe.redis.console.healthcheck.ping.PingService;
import com.ctrip.xpipe.utils.DateTimeUtils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author chen.zhu
 * <p>
 * Sep 06, 2018
 */
public class DelayAction extends AbstractHealthCheckAction<DelayActionContext> {

    public static final String CHECK_CHANNEL = "xpipe-health-check-" + FoundationService.DEFAULT.getLocalIp();

    private static final DelayActionContext INIT_CONTEXT = new DelayActionContext(null, HealthStatus.UNSET_TIME);

    public static final long SAMPLE_LOST_AND_NO_PONG = -99999L * 1000 * 1000;

    public static final long SAMPLE_LOST_BUT_PONG = 99999L * 1000 * 1000;

    private SubscribeCallback callback = new SubscribeCallback();

    private AtomicReference<DelayActionContext> context = new AtomicReference<>(INIT_CONTEXT);

    private PingService pingService;

    private final long expireInterval;

    public DelayAction(ScheduledExecutorService scheduled, RedisHealthCheckInstance instance,
                       ExecutorService executors, PingService pingService) {
        super(scheduled, instance, executors);
        this.pingService = pingService;
        expireInterval = instance.getHealthCheckConfig().getHealthyDelayMilli() + DELTA * 2;
    }

    @Override
    protected void doScheduledTask() {
        reportDelay();
        RedisSession session = instance.getRedisSession();
        session.subscribeIfAbsent(CHECK_CHANNEL, callback);
        if(instance.getRedisInstanceInfo().isMaster()) {
            session.publish(CHECK_CHANNEL, Long.toHexString(System.nanoTime()));
        }
    }

    private void reportDelay() {
        if(INIT_CONTEXT.equals(context.get())) {
            return;
        }
        if(isExpired()) {
            logger.warn("[expire][{}] last update time: {}", instance.getRedisInstanceInfo().getHostPort(),
                    DateTimeUtils.timeAsString(TimeUnit.NANOSECONDS.toMillis(context.get().getRecvTimeNano())));

            long result = SAMPLE_LOST_AND_NO_PONG;
            if(pingService.isRedisAlive(instance.getRedisInstanceInfo().getHostPort())) {
                result = SAMPLE_LOST_BUT_PONG;
            }
            notifyListeners(new DelayActionContext(instance, result));
        } else {
            notifyListeners(context.get());
        }
    }

    private class SubscribeCallback implements RedisSession.SubscribeCallback {

        @Override
        public void message(String channel, String message) {
            onMessage(message);
        }

        @Override
        public void fail(Throwable e) {
            //ignore sub fail
        }
    }

    private void onMessage(String message) {
        long currentTime = System.nanoTime();
        long lastDelayPubTimeNano = Long.parseLong(message, 16);
        this.context.set(new DelayActionContext(instance, currentTime - lastDelayPubTimeNano));
    }

    private boolean isExpired() {
        long lastDelay = System.nanoTime() - context.get().getRecvTimeNano();
        return TimeUnit.NANOSECONDS.toMillis(lastDelay) >= expireInterval;
    }

}
{'fraction_non_alphanumeric': 0.09570552147239264, 'fraction_numerical': 0.00954328561690525, 'mean_word_length': 3.4219409282700424, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4269415', 'n_tokens_mistral': 2155, 'n_tokens_neox': 2034, 'n_words': 368}