<commit_before>/*
 *
 *  Copyright 2016 <NAME>
 *
 *  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.
 *
 *
 */
package io.github.robwin.retry.internal;

import io.github.robwin.retry.Retry;
import io.github.robwin.retry.RetryConfig;
import io.github.robwin.retry.event.RetryEvent;
import io.github.robwin.retry.event.RetryOnErrorEvent;
import io.github.robwin.retry.event.RetryOnSuccess;
import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import javaslang.collection.Stream;
import javaslang.control.Option;
import javaslang.control.Try;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;

public class RetryContext implements Retry {

    private final AtomicInteger numOfAttempts = new AtomicInteger(0);
    private AtomicReference<Exception> lastException = new AtomicReference<>();
    private AtomicReference<RuntimeException> lastRuntimeException = new AtomicReference<>();

    private final FlowableProcessor<RetryEvent> eventPublisher;

    private String id;
    private int maxAttempts;
    private Duration waitDuration;
    private Function<Duration, Duration> backoffFunction;
    private Predicate<Throwable> exceptionPredicate;
    /*package*/ static Try.CheckedConsumer<Long> sleepFunction = Thread::sleep;

    public RetryContext(String id, RetryConfig config){
        this.id = id;
        this.maxAttempts = config.getMaxAttempts();
        this.waitDuration = config.getWaitDuration();
        this.backoffFunction = config.getBackoffFunction();
        this.exceptionPredicate = config.getExceptionPredicate();
        this.eventPublisher = PublishProcessor.create();
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public void onSuccess() {
        int currentNumOfAttempts = numOfAttempts.get();
        if(currentNumOfAttempts > 0){
            Throwable throwable = Option.of(lastException.get()).getOrElse(lastRuntimeException.get());
            publishRetryEvent(() -> new RetryOnSuccess(getId(), currentNumOfAttempts, throwable));
        }
    }

    private void throwOrSleepAfterException() throws Exception {
        int currentNumOfAttempts = numOfAttempts.incrementAndGet();
        if(currentNumOfAttempts == maxAttempts){
            Exception throwable = lastException.get();
            publishRetryEvent(() -> new RetryOnErrorEvent(getId(), currentNumOfAttempts, throwable));
            throw throwable;
        }else{
            waitIntervalAfterFailure();
        }
    }

    public void throwOrSleepAfterRuntimeException(){
        int currentNumOfAttempts = numOfAttempts.incrementAndGet();
        if(currentNumOfAttempts == maxAttempts){
            RuntimeException throwable = lastRuntimeException.get();
            publishRetryEvent(() -> new RetryOnErrorEvent(getId(), currentNumOfAttempts, throwable));
            throw throwable;
        }else{
            waitIntervalAfterFailure();<commit_msg>Fix for race conditions if two threads increase the max error counter at the same time.<commit_after>/*
 *
 *  Copyright 2016 <NAME>
 *
 *  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.
 *
 *
 */
package io.github.robwin.retry.internal;

import io.github.robwin.retry.Retry;
import io.github.robwin.retry.RetryConfig;
import io.github.robwin.retry.event.RetryEvent;
import io.github.robwin.retry.event.RetryOnErrorEvent;
import io.github.robwin.retry.event.RetryOnSuccess;
import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import javaslang.collection.Stream;
import javaslang.control.Option;
import javaslang.control.Try;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;

public class RetryContext implements Retry {

    private final AtomicInteger numOfAttempts = new AtomicInteger(0);
    private AtomicReference<Exception> lastException = new AtomicReference<>();
    private AtomicReference<RuntimeException> lastRuntimeException = new AtomicReference<>();

    private final FlowableProcessor<RetryEvent> eventPublisher;

    private String id;
    private int maxAttempts;
    private Duration waitDuration;
    private Function<Duration, Duration> backoffFunction;
    private Predicate<Throwable> exceptionPredicate;
    /*package*/ static Try.CheckedConsumer<Long> sleepFunction = Thread::sleep;

    public RetryContext(String id, RetryConfig config){
        this.id = id;
        this.maxAttempts = config.getMaxAttempts();
        this.waitDuration = config.getWaitDuration();
        this.backoffFunction = config.getBackoffFunction();
        this.exceptionPredicate = config.getExceptionPredicate();
        this.eventPublisher = PublishProcessor.create();
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public void onSuccess() {
        int currentNumOfAttempts = numOfAttempts.get();
        if(currentNumOfAttempts > 0){
            Throwable throwable = Option.of(lastException.get()).getOrElse(lastRuntimeException.get());
            publishRetryEvent(() -> new RetryOnSuccess(getId(), currentNumOfAttempts, throwable));
        }
    }

    private void throwOrSleepAfterException() throws Exception {
        int currentNumOfAttempts = numOfAttempts.incrementAndGet();
        if(currentNumOfAttempts >= maxAttempts){
            Exception throwable = lastException.get();
            publishRetryEvent(() -> new RetryOnErrorEvent(getId(), currentNumOfAttempts, throwable));
            throw throwable;
        }else{
            waitIntervalAfterFailure();
        }
    }

    public void throwOrSleepAfterRuntimeException(){
        int currentNumOfAttempts = numOfAttempts.incrementAndGet();
        if(currentNumOfAttempts >= maxAttempts){
            RuntimeException throwable = lastRuntimeException.get();
            publishRetryEvent(() -> new RetryOnErrorEvent(getId(), currentNumOfAttempts, throwable));
            throw throwable;
        }else{
            waitIntervalAfterFailure();
{'fraction_non_alphanumeric': 0.08991935483870968, 'fraction_numerical': 0.002688172043010753, 'mean_word_length': 3.7244444444444444, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5935515', 'n_tokens_mistral': 2026, 'n_tokens_neox': 1963, 'n_words': 506}