package es.moki.ratelimitj.redis.request;


import io.lettuce.core.api.StatefulRedisConnection;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.stream.Collectors;

import static java.util.Objects.requireNonNull;

public class RedisScriptLoader {

    private final StatefulRedisConnection<String, String> connection;
    private final String scriptUri;
    private final Flux<StoredScript> storedScript;

    private Disposable cacheDisposable;

    public RedisScriptLoader(StatefulRedisConnection<String, String> connection, String scriptUri) {
        this(connection, scriptUri, false);
    }

    public RedisScriptLoader(StatefulRedisConnection<String, String> connection, String scriptUri, boolean eagerLoad) {
        requireNonNull(connection);
        this.connection = connection;
        this.scriptUri = requireNonNull(scriptUri);

        this.storedScript = Flux
                .defer(this::loadScript)
                .replay(1)
                .autoConnect(1, disposable -> {
                    this.cacheDisposable = disposable;
                });

        if (eagerLoad) {
            this.storedScript.blockFirst(Duration.ofSeconds(10));
        }
    }

    Mono<StoredScript> storedScript() {
        return storedScript.next();
    }

    private Mono<StoredScript> loadScript() {
        String script;
        try {
            script = readScriptFile();
        } catch (IOException e) {
            return Mono.error(new RuntimeException("Unable to load Redis LUA script file", e));
        }

        return connection.reactive()
                .scriptLoad(script)
                .map(sha -> new StoredScript(sha, cacheDisposable));
    }

    private String readScriptFile() throws IOException {
        URL url = RedisScriptLoader.class.getClassLoader().getResource(scriptUri);

        if (url == null) {
            throw new IllegalArgumentException("script '" + scriptUri + "' not found");
        }

        try (BufferedReader reader = new BufferedReader(new InputStreamReader(url.openStream(), StandardCharsets.UTF_8))) {
            return reader.lines().collect(Collectors.joining("\n"));
        }
    }

    static class StoredScript implements Disposable {
        private String sha;

        private Disposable disposable;

        StoredScript(String sha, Disposable disposable) {
            this.sha = sha;
            this.disposable = disposable;
        }

        public String getSha() {
            return sha;
        }

        @Override
        public boolean isDisposed() {
            return disposable.isDisposed();
        }

        @Override
        public void dispose() {
            disposable.dispose();
        }
    }

}

@@ -93,40 +93,8 @@
 on;%0A
-import reactor.core.Disposable;%0A
 impo
@@ -341,24 +341,76 @@
 e.Duration;%0A
+import java.util.concurrent.atomic.AtomicReference;%0A
 import java.
@@ -645,80 +645,50 @@
 nal 
-Flux%3CStoredScript%3E storedScript;%0A%0A    private Disposable cacheDisposable
+AtomicReference%3CFlux%3CString%3E%3E storedScript
 ;%0A%0A 
@@ -1117,201 +1117,42 @@
 t = 
-Flux%0A                .defer(this::loadScript)%0A                .replay(1)%0A                .autoConnect(1, disposable -%3E %7B%0A                    this.cacheDisposable = disposable;%0A                %7D
+new AtomicReference%3C%3E(loadScript()
 );%0A%0A
@@ -1206,16 +1206,22 @@
 dScript.
+get().
 blockFir
@@ -1324,26 +1324,171 @@
 urn 
-storedScript.next(
+Mono.defer(() -%3E %7B%0A            Flux%3CString%3E source = this.storedScript.get();%0A            return source.next().map(sha -%3E new StoredScript(sha, source));%0A        %7D
 );%0A 
@@ -1505,33 +1505,27 @@
 private 
-Mono%3CStoredScript
+Flux%3CString
 %3E loadSc
@@ -1525,32 +1525,70 @@
  loadScript() %7B%0A
+        return Flux.defer(() -%3E %7B%0A    
         String s
@@ -1598,24 +1598,28 @@
 pt;%0A        
+    
 try %7B%0A      
@@ -1624,16 +1624,20 @@
         
+   
+ 
 script =
@@ -1651,24 +1651,28 @@
 riptFile();%0A
+    
         %7D ca
@@ -1705,27 +1705,31 @@
         
+    
 return 
-Mono
+Flux
 .error(n
@@ -1793,35 +1793,43 @@
 %22, e));%0A        
-%7D%0A%0A
+    %7D%0A%0A    
         return c
@@ -1848,33 +1848,16 @@
 active()
-%0A                
 .scriptL
@@ -1867,16 +1867,17 @@
 (script)
+;
 %0A       
@@ -1881,66 +1881,34 @@
     
-        .map(sha -%3E new StoredScript(sha, cacheDisposable)
+%7D).replay(1).autoConnect(1
 );%0A 
@@ -2394,23 +2394,16 @@
  %7D%0A%0A    
-static 
 class St
@@ -2417,30 +2417,8 @@
 ipt 
-implements Disposable 
 %7B%0A  
@@ -2460,37 +2460,37 @@
 private 
-Disposable disposable
+Flux%3CString%3E expected
 ;%0A%0A     
@@ -2521,29 +2521,29 @@
 ha, 
-Disposable disposable
+Flux%3CString%3E expected
 ) %7B%0A
@@ -2591,31 +2591,27 @@
 his.
-disposable = disposable
+expected = expected
 ;%0A  
@@ -2699,51 +2699,27 @@
     
-@Override%0A        public boolean isD
+public void d
 ispose
-d
 () %7B
@@ -2735,39 +2735,39 @@
     
-return disposable.isDisposed();
+storedScript.weakCompareAndSet(
 %0A   
@@ -2771,27 +2771,24 @@
         
-%7D%0A%0A
         
 @Overrid
@@ -2783,81 +2783,67 @@
     
-@Override%0A        public void dispose() %7B%0A            disposable.dispose(
+    expected,%0A                    loadScript()%0A            
 );%0A 

use AtomicReference