/*
 * 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.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
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 JedisPool jedisPool;

	// 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);
		JedisPoolConfig config = new JedisPoolConfig();
		// TODO: make these configurable
		config.setMaxWaitMillis(100);
		config.setMaxTotal(2);
		// We prioritize writing the data ASAP so do validation after we attempted the write
		config.setTestOnReturn(true);
		this.jedisPool = new JedisPool(config, this.host, this.port, 0);

		// Try pinging once
		try (Jedis jedis = jedisPool.getResource()){
			if (!jedis.ping().equals("PONG")) {
				throw new IOException("Could not reach redis instance: " + toString());
			}
		}
	}

	@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;
		}

		// Jedis instance will be auto-returned to the pool
		try (Jedis jedis = jedisPool.getResource()){
			// Test connectivity, this is cheaper than a full validation (i.e. ping)
			if (!jedis.isConnected()) {
				jedis.connect();
			}

			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;
		// Jedis instance will be auto-returned to the pool
		try (Jedis jedis = jedisPool.getResource()){
			// Test connectivity, this is cheaper than a full validation (i.e. ping)
			if (!jedis.isConnected()) {
				jedis.connect();
			}

			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.jedisPool.destroy();
	}

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

@@ -302,84 +302,84 @@
 ava.
-util.concurrent.TimeoutException;%0A%0Aimport com.google.common.io.BaseEncoding;
+nio.charset.StandardCharsets;%0Aimport java.util.concurrent.TimeoutException;%0A
 %0Aimp
@@ -2327,54 +2327,90 @@
 (key
-, subkey, BaseEncoding.base64().encode(
+.getBytes(StandardCharsets.UTF_8), subkey.getBytes(StandardCharsets.UTF_8), 
 value)
-)
 ;%0A%09%09
@@ -2891,22 +2891,22 @@
 %0A%09%09%7D%0A%0A%09%09
-String
+byte%5B%5D
  res;%0A%09%09
@@ -3161,24 +3161,90 @@
 hget(key
-, subkey
+.getBytes(StandardCharsets.UTF_8), subkey.getBytes(StandardCharsets.UTF_8)
 );%0A%09%09%7D c
@@ -3401,41 +3401,11 @@
 urn 
-BaseEncoding.base64().decode(
 res
-)
 ;%0A%09%7D

Use byte arrays for jedis client calls