<commit_before>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();
  }

}<commit_msg>Implement first bits of redis state provider<commit_after>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;
  }

  
}
{'fraction_non_alphanumeric': 0.11625708884688091, 'fraction_numerical': 0.0023629489603024575, 'mean_word_length': 3.320408163265306, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5529936', 'n_tokens_mistral': 663, 'n_tokens_neox': 655, 'n_words': 149}