package be.wegenenverkeer.atomium.client;

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Single;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.awt.*;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class AtomiumFeed<E> {
    private static String FIRST_PAGE = "/";
    private final static Logger logger = LoggerFactory.getLogger(AtomiumFeed.class);
    private final PageFetcher<E> pageFetcher;

    public AtomiumFeed(PageFetcher<E> pageFetcher) {
        this.pageFetcher = pageFetcher;
    }

    private int retryCount = 0;

    public Flowable<FeedEntry<E>> fetchEntries(FeedPositionStrategy feedPositionStrategy) {
        // recursion breaks backpressure, so we're using a generator:
        AtomicReference<CachedFeedPage<E>> previousPage = new AtomicReference<>(null);
        return Flowable.generate( // generate an infinite stream of zeroes (Flowable.range only goes to Integer.MAX_VALUE)
                () -> 0,
                (s, emitter) -> {
                    emitter.onNext(0);
                }
            ).concatMap(t -> { // fetch the page to pass to the feed position strategy
                    Single<CachedFeedPage<E>> page;
                    if (previousPage.get() != null) {
                        page = Single.just(previousPage.get());
                    } else {
                        page = fetchHeadPage();
                    }
                    return page.flatMap(eCachedFeedPage -> feedPositionStrategy.getNextFeedPosition(eCachedFeedPage) // get next feed position
                                    .flatMap(feedPosition -> fetchPage(feedPosition.pageUrl, eCachedFeedPage.etag) // fetch entries, and parse them
                                            .doOnSuccess(previousPage::set)
                                            .map(cachedFeedPage -> ParsedFeedPage.parse(cachedFeedPage, feedPosition))))
                            .toFlowable()
                            .flatMap(parsedPage -> Flowable.fromIterable(parsedPage.getEntries()));
                }, 1);
    }

    private Single<CachedFeedPage<E>> fetchHeadPage() {
        return fetchPage(AtomiumFeed.FIRST_PAGE, Optional.empty());
    }

    private Single<CachedFeedPage<E>> fetchPage(String pageUrl, Optional<String> eTag) {
        return pageFetcher.fetch(pageUrl, eTag)
                .retryWhen(throwableFlowable -> throwableFlowable
                        .flatMap(this::applyRetryStrategy)
                        .flatMap(delay -> Flowable.just("ignored").delay(delay.longValue(), TimeUnit.MILLISECONDS))
                )
                .doAfterSuccess(page -> this.retryCount = 0);
    }

    private Flowable<Long> applyRetryStrategy(Throwable throwable) {
        try {
            return Flowable.just(pageFetcher.getRetryStrategy().apply(++this.retryCount, throwable));
        } catch(Throwable e) {
            return Flowable.error(e);
        }
    }
}

@@ -36,16 +36,65 @@
 lient;%0A%0A
+import io.reactivex.rxjava3.annotations.NonNull;%0A
 import i
@@ -165,24 +165,62 @@
 ore.Single;%0A
+import org.reactivestreams.Publisher;%0A
 import org.s
@@ -837,78 +837,8 @@
 ) %7B%0A
-        // recursion breaks backpressure, so we're using a generator:%0A
     
@@ -920,16 +920,17 @@
 (null);%0A
+%0A
         
@@ -940,639 +940,435 @@
 urn 
-Flowable.generate( // generate an infinite stream of zeroes (Flowable.range only goes to Integer.MAX_VALUE)%0A                () -%3E 0,%0A                (s, emitter) -%3E %7B%0A                    emitter.onNext(0);%0A                %7D%0A            ).concatMap(t -%3E %7B // fetch the page to pass to the
+infiniteZeroes() // recursion breaks backpressure, so we're using a generator of infinite zeroes%0A                .concatMap(t -%3E fetchEntries(feedPositionStrategy, previousPage), 1);%0A    %7D%0A%0A    private Publisher%3C? extends FeedEntry%3CE%3E%3E fetchEntries(FeedPositionStrategy
  feed
- p
+P
 osition
- s
+S
 trategy
-%0A                    Single%3CCachedFeedPage%3CE%3E%3E page;%0A                    if (previousPage.get() != null) %7B%0A                        page = Single.just(previousPage.get());%0A                    %7D else %7B%0A                        page = fetchHeadPage();%0A                    %7D%0A                    return page.flatMap(eCachedFeed
+, AtomicReference%3CCachedFeedPage%3CE%3E%3E previousPageRef) %7B%0A        return getPreviousPage(previousPageRef)%0A                .flatMap(previous
 Page
@@ -1412,27 +1412,24 @@
 osition(
-eCachedFeed
+previous
 Page) //
@@ -1472,36 +1472,24 @@
             
-    
-        
 .flatMap(fee
@@ -1537,19 +1537,16 @@
 rl, 
-eCachedFeed
+previous
 Page
@@ -1572,24 +1572,8 @@
 ries
-, and parse them
 %0A   
@@ -1601,28 +1601,16 @@
         
-            
 .doOnSuc
@@ -1630,27 +1630,78 @@
 Page
+Ref
 ::set)
-%0A            
+ // update ref so we're processing the correct page later on%0A
     
@@ -1804,21 +1804,34 @@
 tion))))
-%0A    
+ // parse entries%0A
         
@@ -1838,24 +1838,16 @@
         
-        
 .toFlowa
@@ -1852,28 +1852,16 @@
 wable()%0A
-            
         
@@ -1944,28 +1944,592 @@
 ));%0A    
+%7D%0A%0A
     
+private Single%3CCachedFeedPage%3CE%3E%3E getPreviousPage(AtomicReference%3CCachedFeedPage%3CE%3E%3E previousPage) %7B%0A        if (previousPage.get() != null) %7B%0A            return Single.just(previousPage.get());%0A        %7D else %7B%0A            return fetchHeadPage();%0A        %7D%0A    %7D%0A%0A    // generate an infinite stream of zeroes (Flowable.range only goes to Integer.MAX_VALUE)%0A    private Flowable%3CInteger%3E infiniteZeroes() %7B%0A        return Flowable.generate(%0A                () -%3E 0,%0A                (s, emitter) -%3E %7B%0A                    emitter.onNext(0);%0A                %7D%0A    
         
 %7D, 1);%0A 
@@ -2524,12 +2524,8 @@
     
-%7D, 1
 );%0A 

improve readability