package reactor.pipe.redis;

import reactor.pipe.codec.Codec;
import reactor.pipe.concurrent.Atom;
import reactor.pipe.state.StateProvider;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import redis.clients.jedis.*;

public class RedisStateProvider implements StateProvider {

  private final Map<Class, Codec>        encoders;
  private final ScheduledExecutorService executor;
  private final JedisPool                pool;

  public RedisStateProvider() {
    this.encoders = new HashMap<>();
    executor = Executors.newScheduledThreadPool(1);
    pool = new JedisPool(new JedisPoolConfig(),
                         "localhost",
                         6379);
  }

  public <T> void registerCodec(Class<T> klass, Codec<T, String> encoder) {
    this.encoders.put(klass, encoder);
  }

  @SuppressWarnings("unchecked")
  protected <IN, OUT> Codec<IN, OUT> getCodec(Class<IN> klass) {
    return (Codec<IN, OUT>) this.encoders.get(klass);
  }

  @Override
  public <SRC, T> Atom<T> makeAtom(SRC src, T init) { // Init should never be null!
    Codec<SRC, String> keyCodec = getCodec((Class<SRC>) init.getClass());
    Codec<T, String> valueCodec = getCodec((Class<T>) init.getClass());

    Atom<T> atom = new Atom<>(init);

    executor.submit(new Runnable() {
      @Override
      public void run() {
        try (Jedis client = pool.getResource()) {
          client.set(keyCodec.encode(src),
                     valueCodec.encode(atom.deref()));
        } catch (Exception e) {
          //
        }

      }
    });

    return atom;
  }

  
}

@@ -1077,16 +1077,49 @@
 verride%0A
+  @SuppressWarnings(%22unchecked%22)%0A
   public
@@ -1250,20 +1250,19 @@
 s%3CSRC%3E) 
-init
+src
 .getClas
@@ -1374,16 +1374,34 @@
 m%3C%3E(init
+ialLoad(src, init)
 );%0A%0A    
@@ -1718,13 +1718,512 @@
 m;%0A  %7D%0A%0A
-
   
+@SuppressWarnings(%22unchecked%22)%0A  protected %3CSRC, T%3E T initialLoad(SRC src, T init) %7B%0A    Codec%3CSRC, String%3E keyCodec = getCodec((Class%3CSRC%3E) src.getClass());%0A    Codec%3CT, String%3E valueCodec = getCodec((Class%3CT%3E) init.getClass());%0A%0A    try (Jedis client = pool.getResource()) %7B%0A      String loaded = client.get(keyCodec.encode(src));%0A      if (loaded == null) %7B%0A        return init;%0A      %7D else %7B%0A        return valueCodec.decode(loaded);%0A      %7D%0A    %7D catch (Exception e) %7B%0A      throw e;%0A    %7D%0A  %7D
 %0A%7D%0A

Fix initial loading