package reactor.pipe.redis;

import reactor.pipe.concurrent.Atom;
import reactor.pipe.state.StateProvider;
import sun.reflect.generics.reflectiveObjects.NotImplementedException;

public class RedisStateProvider implements StateProvider {

  @Override
  public <SRC, T> Atom<T> makeAtom(SRC src, T init) {
    throw new NotImplementedException();
  }

}

@@ -22,16 +22,49 @@
 redis;%0A%0A
+import reactor.pipe.codec.Codec;%0A
 import r
@@ -137,250 +137,1544 @@
 er;%0A
+%0A
 import 
-sun.reflect.generics.reflectiveObjects.NotImplementedException;%0A%0Apublic class RedisStateProvider implements StateProvider %7B%0A%0A  @Override%0A  public %3CSRC, T%3E Atom%3CT%3E makeAtom(SRC src, T init) %7B%0A    throw new NotImplementedException()
+java.nio.ByteBuffer;%0Aimport java.util.HashMap;%0Aimport java.util.Map;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.ScheduledExecutorService;%0Aimport redis.clients.jedis.*;%0A%0Apublic class RedisStateProvider implements StateProvider %7B%0A%0A  private final Map%3CClass, Codec%3E        encoders;%0A  private final ScheduledExecutorService executor;%0A  private final JedisPool                pool;%0A%0A  public RedisStateProvider() %7B%0A    this.encoders = new HashMap%3C%3E();%0A    executor = Executors.newScheduledThreadPool(1);%0A    pool = new JedisPool(new JedisPoolConfig(),%0A                         %22localhost%22,%0A                         6379);%0A  %7D%0A%0A  public %3CT%3E void registerCodec(Class%3CT%3E klass, Codec%3CT, String%3E encoder) %7B%0A    this.encoders.put(klass, encoder);%0A  %7D%0A%0A  @SuppressWarnings(%22unchecked%22)%0A  protected %3CIN, OUT%3E Codec%3CIN, OUT%3E getCodec(Class%3CIN%3E klass) %7B%0A    return (Codec%3CIN, OUT%3E) this.encoders.get(klass);%0A  %7D%0A%0A  @Override%0A  public %3CSRC, T%3E Atom%3CT%3E makeAtom(SRC src, T init) %7B // Init should never be null!%0A    Codec%3CSRC, String%3E keyCodec = getCodec((Class%3CSRC%3E) init.getClass());%0A    Codec%3CT, String%3E valueCodec = getCodec((Class%3CT%3E) init.getClass());%0A%0A    Atom%3CT%3E atom = new Atom%3C%3E(init);%0A%0A    executor.submit(new Runnable() %7B%0A      @Override%0A      public void run() %7B%0A        try (Jedis client = pool.getResource()) %7B%0A          client.set(keyCodec.encode(src),%0A                     valueCodec.encode(atom.deref()));%0A        %7D catch (Exception e) %7B%0A          //%0A        %7D%0A%0A      %7D%0A    %7D);%0A%0A    return atom
 ;%0A  %7D%0A%0A
+  %0A
 %7D%0A

Implement first bits of redis state provider