/*
 * Copyright (C) 2016 Turn Inc. All Rights Reserved.
 * Proprietary and confidential.
 */

package com.turn.edc.storage.impl;

import com.turn.edc.exception.KeyNotFoundException;
import com.turn.edc.storage.StorageConnector;

import java.io.IOException;
import java.lang.ref.WeakReference;
import java.util.concurrent.TimeoutException;

import com.google.common.io.BaseEncoding;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

/**
 * ConnectionFactory connector to Redis using Jedis library
 * (https://github.com/xetorthio/jedis)
 *
 * @author tshiou
 */
public class JedisStorageConnector extends StorageConnector {

	private static final String DEFAULT_SUBKEY = "_SINGLEFIELD";

	private final String host;
	private final int port;
	private final Jedis jedis;

	// Weak reference since most of the time we don't need the string representation
	private WeakReference<String> toString = new WeakReference<>(null);

	public JedisStorageConnector(String host, String port, int timeout) throws IOException {

		this.host = host;
		this.port = Integer.parseInt(port);
		this.jedis = new Jedis(this.host, this.port, 0);

		try {
			this.jedis.ping();
		} catch (Exception e) {
			throw new IOException(e);
		}
	}

	@Override
	public void set(String key, byte[] value, int ttl, int timeout) throws IOException {
		set(key, DEFAULT_SUBKEY, value, ttl, timeout);
	}

	@Override
	public void set(String key, String subkey, byte[] value, int ttl, int timeout) throws IOException {
		if (subkey == null || subkey.isEmpty()) {
			subkey = DEFAULT_SUBKEY;
		}

		try {
			Pipeline p = jedis.pipelined();
			p.hset(key, subkey, BaseEncoding.base64().encode(value));
			p.expire(key, ttl);
			p.sync();
		} catch (Exception e) {
			throw new IOException(e.getCause());
		}
	}

	@Override
	public byte[] get(String key, int timeout) throws KeyNotFoundException, TimeoutException, IOException {
		return get(key, DEFAULT_SUBKEY, timeout);
	}

	@Override
	public byte[] get(String key, String subkey, int timeout) throws KeyNotFoundException, TimeoutException, IOException {
		if (subkey == null || subkey.isEmpty()) {
			subkey = DEFAULT_SUBKEY;
		}

		String res;
		try {
			res = jedis.hget(key, subkey);
		} catch (Exception e) {
			throw new IOException(e.getCause());
		}
		if (res == null) {
			throw new KeyNotFoundException(key + ":" + subkey);
		}
		return BaseEncoding.base64().decode(res);
	}

	@Override
	public void close() {
		this.jedis.close();
	}

	@Override
	public String toString() {
		if (this.toString.get() == null) {
			this.toString = new WeakReference<String>(
					(new StringBuilder())
							.append("Jedis ")
							.append(host)
							.append(":")
							.append(port)
							.toString()
			);
		}
		return this.toString.get();
	}
}

@@ -409,16 +409,98 @@
 .Jedis;%0A
+import redis.clients.jedis.JedisPool;%0Aimport redis.clients.jedis.JedisPoolConfig;%0A
 import r
@@ -857,22 +857,30 @@
 al Jedis
+Pool
  jedis
+Pool
 ;%0A%0A%09// W
@@ -1175,16 +1175,277 @@
 (port);%0A
+%09%09JedisPoolConfig config = new JedisPoolConfig();%0A%09%09// TODO: make these configurable%0A%09%09config.setMaxWaitMillis(100);%0A%09%09config.setMaxTotal(2);%0A%09%09// We prioritize writing the data ASAP so do validation after we attempted the write%0A%09%09config.setTestOnReturn(true);%0A
 %09%09this.j
@@ -1448,16 +1448,20 @@
 is.jedis
+Pool
  = new J
@@ -1464,17 +1464,29 @@
 ew Jedis
-(
+Pool(config, 
 this.hos
@@ -1510,62 +1510,115 @@
 %0A%0A%09%09
-try %7B%0A%09%09%09this.
+// Try pinging once%0A%09%09try (Jedis 
 jedis
-.ping();%0A%09%09%7D catch (Exception e
+ = jedisPool.getResource())%7B%0A%09%09%09if (!jedis.ping().equals(%22PONG%22)
 ) %7B%0A
+%09
 %09%09%09t
@@ -1638,19 +1638,70 @@
 ception(
-e);
+%22Could not reach redis instance: %22 + toString());%0A%09%09%09%7D
 %0A%09%09%7D%0A%09%7D%0A
@@ -2042,21 +2042,248 @@
 %0A%09%09%7D%0A%0A%09%09
-try %7B
+// Jedis instance will be auto-returned to the pool%0A%09%09try (Jedis jedis = jedisPool.getResource())%7B%0A%09%09%09// Test connectivity, this is cheaper than a full validation (i.e. ping)%0A%09%09%09if (!jedis.isConnected()) %7B%0A%09%09%09%09jedis.connect();%0A%09%09%09%7D%0A
 %0A%09%09%09Pipe
@@ -2873,13 +2873,240 @@
 ;%0A%09%09
-try %7B
+// Jedis instance will be auto-returned to the pool%0A%09%09try (Jedis jedis = jedisPool.getResource())%7B%0A%09%09%09// Test connectivity, this is cheaper than a full validation (i.e. ping)%0A%09%09%09if (!jedis.isConnected()) %7B%0A%09%09%09%09jedis.connect();%0A%09%09%09%7D%0A
 %0A%09%09%09
@@ -3384,14 +3384,20 @@
 edis
-.close
+Pool.destroy
 ();%0A

Use JediPool for redis connections