public class RequestContext {

    private DeadlineManager dm;

    private String reqId = null;
    private int reqStart = -1;
    private int rawTimeout = -1;
    private int deadline = -1;

    /**
     * Build a new {@link RequestContext} with a default 30s timeout.
     */
    public RequestContext() {
        this.dm = DeadlineManager.instance();
    }

    /**
     * Copy constructor. Build a new {@link RequestContext} from another one.
     * This will keep the request ID, the timeout and the deadline (if there is one).
     *
     * @param src The {@link RequestContext} to copy.
     */
    public RequestContext(RequestContext src) {
        this.withRequestId(src.requestId());
        this.deadline = src.deadline;
        this.rawTimeout = src.rawTimeout;
    }

    /* -- Request IDs ----------------------------------------------------- */

    /**
     * Ensure this request has an ID and it is at least 8 characters.
     *
     * @return this
     */
    public RequestContext ensureRequestId() {
        if (this.reqId == null)
            this.reqId = IdGen.requestId();
        else if (this.reqId.length() < 8)
            this.reqId += IdGen.requestId().substring(8 - this.reqId.length());
        return this;
    }

    /**
     * Get the request ID.
     *
     * @return the request ID (auto-generated if not set)
     */
    public String requestId() {
        return this.ensureRequestId().reqId;
    }

    /**
     * Set a request ID.
     *
     * @param requestId
     *            a request ID string. If is null, it will be auto-generated. If
     *            it is less than 8 characters, it will be suffixed.
     * @return this
     */
    public RequestContext withRequestId(String requestId) {
        this.reqId = requestId;
        return this;
    }

    /* -- Deadlines and timeouts ------------------------------------------ */

    /**
     * Compute a deadline from the {@link #timeout()}. This also means that the
     * timeout will be recomputed for each unique sub-request resulting from an
     * API call. If this request has started, compute the timeout from the start,
     * otherwise compute it from now.
     *
     * @return {@code this}
     */
    public RequestContext computeDeadline() {
        if (!this.hasTimeout())
            throw new IllegalStateException("No timeout has been set, cannot compute deadline");
        if (this.hasStarted())
            this.deadline = dm.timeoutToDeadline(this.rawTimeout, this.reqStart);
        else
            this.deadline = dm.timeoutToDeadline(this.rawTimeout);
        return this;
    }

    /**
     * Get the deadline for this request.
     *
     * If a timeout has been set ({@link #withTimeout(int)}) but no
     * deadline, {@link #computeDeadline()} will be called.
     *
     * @return the overall deadline for this request, in milliseconds
     */
    public int deadline() {
        if (!hasDeadline()) {
            startTiming();
            computeDeadline();
        }
        return this.deadline;
    }

    /**
     * Tell the time elapsed since this request has started.
     *
     * @return the duration since this request has started, in milliseconds
     */
    public int elapsed() {
        if (!this.hasStarted())
            return 0;
        return this.dm.now() - this.reqStart;
    }

    /**
     * Tell whether this request has a deadline or not. This will always return
     * {@code true} after {@link #computeDeadline()} or {@link #deadline()} has been
     * called.
     *
     * @return {@code true} if this request has a deadline.
     */
    public boolean hasDeadline() {
        return this.deadline >= 0;
    }

    /**
     * Tell if this request has started.
     *
     * @return {@code true} if this request has started.
     */
    public boolean hasStarted() {
        return this.reqStart >= 0;
    }

    /**
     * Tell whether this request has a timeout or not.
     *
     * @return {@code true} if this request has timeout.
     */
    public boolean hasTimeout() {
        return this.rawTimeout >= 0;
    }

    /**
     * Reset the deadline set on this request. This must be called when reusing
     * a context only for its timeout and request ID.
     *
     * @return {@code this}
     */
    public RequestContext resetDeadline() {
        this.deadline = -1;
        return this;
    }

    /**
     * Start timing this request.
     *
     * @return this
     */
    public RequestContext startTiming() {
        this.reqStart = dm.now();
        return this;
    }

    /**
     * Get the timeout for the request.
     *
     * If {@link #hasDeadline()} returns {@code true}, successive calls to this
     * method will return decreasing values, and negative values when the
     * deadline has been exceeded.
     *
     * @return the timeout for this request, in milliseconds
     */
    public int timeout() {
        if (this.hasDeadline())
            return this.dm.deadlineToTimeout(this.deadline);
        return this.rawTimeout;
    }

    /**
     * Set a deadline on the whole request.
     *
     * This will reset any previous timeout set with {@link #withTimeout(int)} to the duration
     * from now to the deadline.
     *
     * @param deadline the deadline in milliseconds
     * @return this
     */
    public RequestContext withDeadline(int deadline) {
        checkArgument(deadline >= 0, "deadline cannot be negative");
        this.deadline = deadline;
        this.rawTimeout = this.dm.deadlineToTimeout(deadline);
        return this;
    }

    /**
     * Set a timeout for each unique sub-request resulting from an API call.
     *
     * This will reset any previous deadline set with {@link #withDeadline(int)}.
     *
     * The timeout will be used to compute the deadline when
     * {@link #startTiming()} or {@link #deadline()} is called.
     *
     * @param timeout
     *            a timeout in milliseconds
     * @return this
     */
    public RequestContext withTimeout(int timeout) {
        checkArgument(timeout > 0, "timeout cannot be negative");
        this.rawTimeout = timeout;
        this.resetDeadline();
        return this;
    }
} public class FeedableInputStream extends InputStream {

    private static final SdsLogger logger = SdsLoggerFactory
            .getLogger(FeedableInputStream.class);

    private LinkedBlockingQueue<DataPart> q;
    private DataPart current = null;
    private boolean failed = false;
    private long pollDelayMillis = 10000;

    /**
     * @param qsize Queue size
     * @param pollDelayMillis Delay between iterations of the read loop
     * @param retries Maximum number of iterations of the read loop
     */
    public FeedableInputStream(int qsize, long pollDelayMillis, int retries) {
        this.q = new LinkedBlockingQueue<FeedableInputStream.DataPart>(qsize);
        this.pollDelayMillis = pollDelayMillis;
    }

    public FeedableInputStream(int qsize, long pollDelayMillis) {
        this(qsize, pollDelayMillis, 3);
    }

    public FeedableInputStream(int qsize) {
        this(qsize, 10000);
    }

    public void setFailed(boolean f) {
        failed = f;
    }

    public boolean isFailed() {
        return failed;
    }

    public void feed(ByteBuffer b, boolean last) {
        if (failed)
            return;
        try {
            DataPart part = new DataPart(b, last);
            while (!(failed || q.offer(part, 1L, TimeUnit.SECONDS))) {
                /* Retry until done
                 * or reader has been interrupted
                 * or we are interrupted */
                logger.debug("Timed out while trying to feed queue, " +
                        "will check if not failed and then retry");
            }
        } catch (InterruptedException e) {
            failed = true;
            logger.warn("feed interrupted", e);
        }
    }

    @Override
    public int read() throws IOException {
        byte[] b = new byte[1];
        int read = read(b, 0, 1);
        return -1 == read ? -1 : b[0];
    }

    @Override
    public int read(byte[] buf) throws IOException {
        return read(buf, 0, buf.length);
    }

    @Override
    public int read(byte[] buf, int offset, int length) {
        if (length <= 0)
            return 0;
        int retriesLeft = 5;
        while (current == null && retriesLeft > 0) {
            try {
                current = q.poll(this.pollDelayMillis, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                failed = true;
                current = new DataPart(null, true);
            }
            retriesLeft--;
            if (current == null) {
                logger.warn("Failed to read from client application, " +
                        retriesLeft + " retries left");
            } else if (current.buffer() == null ||
                    !current.buffer().hasRemaining()) {
                if (current.isLast())
                    return -1;
                current = null;
            }
        }

        if (current == null) {
            failed = true;
            return -1;
        }

        int read = Math.min(current.buffer().remaining(),
                Math.min(buf.length - offset, length));
        current.buffer().get(buf, offset, read);

        int rc;
        if (read < length) {
            rc = read + Math.max(0, read(buf, offset + read, length - read));
        } else {
            rc = read;
        }

        if (current.buffer().remaining() == 0)
            current = null;
        return rc;
    }

    public static class DataPart {

        private ByteBuffer buffer;
        private boolean last = false;

        public DataPart(ByteBuffer buffer) {
            this.buffer = buffer;
        }

        public DataPart(ByteBuffer buffer, boolean last) {
            this.buffer = buffer;
            this.last = last;
        }

        public ByteBuffer buffer() {
            return buffer;
        }

        public boolean isLast() {
            return last;
        }

    }

} public class Hex {

	private static final char[] hexArray = "0123456789ABCDEF".toCharArray();

	private byte[] wrapped;
	private String str;

	private Hex(byte[] wrapped, String str) {
		this.wrapped = wrapped;
		this.str = str;
	}

	/**
	 * Wraps the specifies byte array
	 * 
	 * @param bytes
	 *            the array to wrap
	 * @return a new Hex wrapping the array
	 * @throws IllegalArgumentException
	 *             if the specified array is {@code null}
	 */
	public static Hex fromBytes(byte[] bytes) {
		Check.checkArgument(null != bytes);
		return new Hex(bytes, toHex(bytes));
	}

	/**
	 * Converts the specified byte array in its hexadecimal string format (in upper case)
	 * @param bytes the byte array to convert
	 * @return the hexadecimal format
	 */
	public static String toHex(byte[] bytes) {
		char[] hexChars = new char[bytes.length * 2];
		for (int j = 0; j < bytes.length; j++) {
			int v = bytes[j] & 0xFF;
			hexChars[j * 2] = hexArray[v >>> 4];
			hexChars[j * 2 + 1] = hexArray[v & 0x0F];
		}
		return new String(hexChars);
	}

	/**
	 * Returns a copy of the wrapped array
	 * @return a copy of the wrapped array
	 */
	public byte[] bytes() {
		return Arrays.copyOf(wrapped, wrapped.length);
	}

	/**
	 * Returns the hexadecimal string format of the wrapped array
	 * @param upperCase specifies if the result will be in upper case or not
	 * @return the hexadecimal string
	 */
	public String hexString(boolean upperCase) {
		return upperCase ? str : str.toLowerCase();
	}

	@Override
	public String toString() {
		return this.str;
	}

} public class Strings {

    private static final SdsLogger LOGGER = SdsLoggerFactory.getLogger(Strings.class);
    public static final String[] EMPTY_ARRAY = new String[0];

    public static boolean nullOrEmpty(String string) {
        return null == string || 0 == string.length();
    }

    public static HashMap<String, String> toMap(String src, String sep,
            String kvsep) {
        HashMap<String, String> res = new HashMap<String, String>();
        if (nullOrEmpty(src) || nullOrEmpty(sep) || nullOrEmpty(kvsep))
            return res;
        String[] pairs = src.split(sep);
        for (String pair : pairs) {
            if (nullOrEmpty(pair))
                continue;
            String[] tok = pair.split(kvsep, 2);
            if (tok.length != 2 || nullOrEmpty(tok[0]) || nullOrEmpty(tok[1]))
                continue;
            res.put(tok[0], tok[1]);
        }
        return res;
    }

    public static List<String> toList(String src, String sep) {
        ArrayList<String> res = new ArrayList<String>();
        if (nullOrEmpty(src) || nullOrEmpty(sep))
            return res;
        String[] toks = src.split(sep);
        for (String tok : toks)
            if (!nullOrEmpty(tok))
                res.add(tok);
        return res;
    }

    public static String urlEncode(String src) {
        try {
            return URLEncoder.encode(src, OioConstants.OIO_CHARSET.name());
        } catch (UnsupportedEncodingException e) {
            LOGGER.warn("Url encoding fail, may have some problems", e);
            return src;
        }
    }

    public static String quote(String src) {
        return Strings.urlEncode(src).replace("+", "%20");
    }

    public static String quote(Long src) {
        return Strings.quote(String.valueOf(src));
    }
} @SuppressWarnings("deprecation")
public class OioException extends SdsException {

    /**
     * 
     */
    private static final long serialVersionUID = 3270900242235005338L;

    public OioException(String message) {
        super(message);
    }

    public OioException(String message, Throwable t) {
        super(message, t);
    }
} public class OioHttp {

    private static final SdsLogger logger = SdsLoggerFactory.getLogger(SdsLogger.class);

    private static final String CRLF = "\r\n";

    private static final byte[] CRLF_BYTES = { '\r', '\n' };

    private OioHttpSettings settings;

    private SocketProvider socketProvider;

    private OioHttp(OioHttpSettings settings, SocketProvider socketProvider) {
        this.settings = settings;
        this.socketProvider = socketProvider;
    }

    public static OioHttp http(OioHttpSettings settings, SocketProvider socketProvider) {
        Check.checkArgument(null != settings);
        Check.checkArgument(null != socketProvider);
        return new OioHttp(settings, socketProvider);
    }

    public RequestBuilder post(String uri) {
        Check.checkArgument(!nullOrEmpty(uri));
        return new RequestBuilder().req(POST_METHOD, uri);
    }

    public RequestBuilder put(String uri) {
        Check.checkArgument(!nullOrEmpty(uri));
        return new RequestBuilder().req(PUT_METHOD, uri);
    }

    public RequestBuilder get(String uri) {
        Check.checkArgument(!nullOrEmpty(uri));
        return new RequestBuilder().req(GET_METHOD, uri);
    }

    public RequestBuilder delete(String uri) {
        Check.checkArgument(!nullOrEmpty(uri));
        return new RequestBuilder().req(DELETE_METHOD, uri);
    }


    public static String timeoutMillisToStringMicros(int timeout) {
        // oio-proxy wants microseconds, and we remove 1% for the parsing overhead.
        return String.valueOf(timeout * 990);
    }

    public class RequestBuilder {

        private String method;
        private HashMap<String, String> headers = new HashMap<String, String>();
        private HashMap<String, String> query = new HashMap<String, String>();
        private String body;
        private InputStream data;
        private Long len;
        private URI uri;
        private OioHttpResponseVerifier verifier = null;
        private boolean chunked;
        private List<InetSocketAddress> hosts = null;
        private RequestContext reqCtx = null;

        public RequestBuilder req(String method, String url) {
            this.method = method;
            this.uri = URI.create(url);
            return this;
        }

        public RequestBuilder header(String name, String value) {
            if (!nullOrEmpty(name) && !nullOrEmpty(value))
                headers.put(name, value);
            return this;
        }

        public RequestBuilder headers(Map<String, String> headers) {
            if (null != headers)
                this.headers.putAll(headers);
            return this;
        }

        public RequestBuilder query(String name, String value) {
            if (!nullOrEmpty(name) && !nullOrEmpty(value))
                this.query.put(name, value);
            return this;
        }

        public RequestBuilder chunked() {
            this.headers.put("Transfer-Encoding", "chunked");
            this.chunked = true;
            return this;
        }

        public RequestBuilder body(String body) {
            if (nullOrEmpty(body))
                return this;
            headers.put(CONTENT_LENGTH_HEADER, String.valueOf(body.length()));
            headers.put(CONTENT_TYPE_HEADER, "application/json");
            this.body = body;
            return this;
        }

        public RequestBuilder body(InputStream data, Long size) {
            if (null == data)
                return this;
            headers.put(CONTENT_LENGTH_HEADER, String.valueOf(size));
            if (!headers.containsKey(CONTENT_TYPE_HEADER)) {
                headers.put(CONTENT_TYPE_HEADER, "application/octet-stream");
            }
            this.data = data;
            this.len = size;
            return this;
        }

        public RequestBuilder verifier(OioHttpResponseVerifier verifier) {
            this.verifier = verifier;
            return this;
        }

        /**
         * Set the context associated to the current request.
         *
         * @param reqCtx
         *            the context to get deadline and request ID from
         * @return {@code this}
         */
        public RequestBuilder withRequestContext(RequestContext reqCtx) {
            this.reqCtx = reqCtx;
            if (reqCtx != null)
                headers.put(OIO_REQUEST_ID_HEADER, reqCtx.requestId());
            return this;
        }

        /**
         * @param hosts
         *            An optional list of hosts to connect to.
         * @return this
         */
        public RequestBuilder hosts(List<InetSocketAddress> hosts) {
            this.hosts = hosts;
            return this;
        }

        public OioHttpResponse execute() throws OioException {
            if (this.hosts == null || this.hosts.isEmpty()) {
                return execute(new InetSocketAddress(uri.getHost(), uri.getPort()));
            } else {
                OioException lastExc = null;
                // TODO: implement better fallback mechanism, with randomization
                for (InetSocketAddress addr : this.hosts) {
                    try {
                        if (lastExc != null)
                            logger.info("Retrying on " + addr.toString());
                        return execute(addr);
                    } catch (OioException oioe) {
                        // Retry only if the cause is network
                        if (oioe.getCause() instanceof IOException) {
                            logger.warn("Failed to perform request", oioe);
                            lastExc = oioe;
                        } else {
                            throw oioe;
                        }
                    }
                }
                throw new OioSystemException("HTTP request execution error", lastExc);
            }
        }

        /**
         * If a deadline has been set on the request:
         * - check it;
         * - set the socket timeout accordingly;
         * Whether a deadline has been set or not, add a timeout header
         * so the server will limit itself (99% of the timeout set on the socket).
         */
        private void applyDeadline(Socket sock) throws SocketException {
            int timeout;
            if (this.reqCtx != null) {
                if (this.reqCtx.hasDeadline())
                    DeadlineManager.instance().checkDeadline(this.reqCtx.deadline());
                if (this.reqCtx.hasDeadline() || this.reqCtx.hasTimeout()) {
                    timeout = this.reqCtx.timeout();
                    sock.setSoTimeout(timeout);
                } else {
                    // Do not change the timeout, but still set the header.
                    timeout = sock.getSoTimeout();
                }
            } else {
                timeout = sock.getSoTimeout();
            }
            headers.put(OIO_TIMEOUT_HEADER, timeoutMillisToStringMicros(timeout));
        }

        private OioHttpResponse execute(InetSocketAddress addr) throws OioException {
            Socket sock = null;
            try {
                sock = socketProvider.getSocket(addr);
                applyDeadline(sock);
                if (chunked)
                    sendRequestChunked(sock);
                else
                    sendRequest(sock);
                OioHttpResponse resp = readResponse(sock);
                try {
                    if (null != verifier)
                        verifier.verify(resp);
                    return resp;
                } catch (OioException e) {
                    resp.close(true);
                    throw e;
                }
            } catch (IOException e) {
                if (null != sock)
                    try {
                        try {
                            sock.shutdownInput();
                        } catch (SocketException se) {
                            logger.debug("Socket input already shutdown");
                        }
                        sock.close();
                    } catch (IOException ioe) {
                        logger.warn("Failed to close socket, possible leak", ioe);
                    }
                throw new OioSystemException("HTTP request execution error", e);
            }
        }

        public <T> T execute(Class<T> c) {
            OioHttpResponse resp = execute();
            boolean success = false;
            try {
                T t = gson().fromJson(
                        new JsonReader(new InputStreamReader(resp.body(), OIO_CHARSET)), c);
                success = true;
                return t;
            } finally {
                resp.close(success);
            }
        }

        private OioHttpResponse readResponse(Socket sock) throws IOException {
            return OioHttpResponse.build(sock, this.reqCtx);
        }

        private void sendRequest(Socket sock) throws IOException {
            headers.put("Host", uri.getHost() + ":" + uri.getPort());
            headers.put("Connection", socketProvider.reusableSocket() ? "keep-alive" : "close");
            headers.put("Accept", "*/*");
            headers.put("Accept-Encoding", "gzip, deflate");
            headers.put("User-Agent", settings.userAgent());

            if (!headers.containsKey("Content-Length"))
                headers.put(CONTENT_LENGTH_HEADER, "0");

            BufferedOutputStream bos = new BufferedOutputStream(sock.getOutputStream(),
                    settings.sendBufferSize());
            bos.write(requestHead());
            if (null != data) {
                stream(bos);
            } else if (null != body) {
                bos.write(body.getBytes(OIO_CHARSET));
            }
            bos.flush();
        }

        private void sendRequestChunked(Socket sock) throws IOException {
            headers.put("Host", uri.getHost() + ":" + uri.getPort());
            headers.put("Connection", socketProvider.reusableSocket() ? "keep-alive" : "close");
            headers.put("Accept", "*/*");
            headers.put("Accept-Encoding", "gzip, deflate");
            headers.put("User-Agent", settings.userAgent());

            // ensure no content-length
            headers.remove("Content-Length");
            BufferedOutputStream bos = new BufferedOutputStream(sock.getOutputStream(),
                    settings.sendBufferSize());
            byte[] requestHead = requestHead();
            bos.write(requestHead);
            streamChunked(bos);
            bos.flush();
        }

        private void streamChunked(OutputStream os) throws IOException {
            byte[] b = new byte[settings.sendBufferSize()];
            int remaining = len.intValue();
            while (remaining > 0) {
                int read = data.read(b, 0, Math.min(remaining, b.length));
                if (-1 == read)
                    throw new EOFException("Unexpected end of source stream");
                remaining -= read;
                if (read > 0) {
                    os.write((Integer.toHexString(read) + CRLF).getBytes(OIO_CHARSET));
                    os.write(b, 0, read);
                    os.write(CRLF_BYTES);
                }
            }
            os.write(("0" + CRLF + CRLF).getBytes(OIO_CHARSET));
        }

        private void stream(OutputStream sos) throws IOException {
            byte[] b = new byte[settings.sendBufferSize()];
            int remaining = len.intValue();

            while (remaining > 0) {
                int read = data.read(b, 0, Math.min(remaining, b.length));
                if (-1 == read)
                    throw new EOFException("Unexpected end of source stream");
                remaining -= read;
                sos.write(b, 0, read);
            }
        }

        private byte[] requestHead() throws UnsupportedEncodingException {
            StringBuilder qbuilder = new StringBuilder(null == uri.getRawQuery() ? ""
                    : uri.getRawQuery());
            boolean removeTrailindAnd = 0 == qbuilder.length();
            for (Entry<String, String> h : query.entrySet()) {
                qbuilder.append("&").append(URLEncoder.encode(h.getKey(), "utf-8")).append("=")
                        .append(URLEncoder.encode(h.getValue(), "utf-8"));
            }
            String uriPath = uri.getRawPath();
            StringBuilder req = new StringBuilder(method)
                    .append(" ")
                    .append(uriPath != null && !uriPath.isEmpty() ? uriPath : "/")
                    .append(qbuilder.length() > 0 ? "?" : "")
                    .append(qbuilder.length() > 0 ? (removeTrailindAnd ? qbuilder.substring(1)
                            : qbuilder.toString()) : "").append(" ").append("HTTP/1.1")
                    .append(CRLF);
            for (Entry<String, String> h : headers.entrySet()) {
                req.append(h.getKey()).append(": ").append(h.getValue()).append(CRLF);
            }
            byte[] res = req.append(CRLF).toString().getBytes(OIO_CHARSET);
            return res;
        }
    }
} public class RequestBuilder {

    private String method;
    private HashMap<String, String> headers = new HashMap<String, String>();
    private HashMap<String, String> query = new HashMap<String, String>();
    private String body;
    private InputStream data;
    private Long len;
    private URI uri;
    private OioHttpResponseVerifier verifier = null;
    private boolean chunked;
    private List<InetSocketAddress> hosts = null;
    private RequestContext reqCtx = null;

    public RequestBuilder req(String method, String url) {
        this.method = method;
        this.uri = URI.create(url);
        return this;
    }

    public RequestBuilder header(String name, String value) {
        if (!nullOrEmpty(name) && !nullOrEmpty(value))
            headers.put(name, value);
        return this;
    }

    public RequestBuilder headers(Map<String, String> headers) {
        if (null != headers)
            this.headers.putAll(headers);
        return this;
    }

    public RequestBuilder query(String name, String value) {
        if (!nullOrEmpty(name) && !nullOrEmpty(value))
            this.query.put(name, value);
        return this;
    }

    public RequestBuilder chunked() {
        this.headers.put("Transfer-Encoding", "chunked");
        this.chunked = true;
        return this;
    }

    public RequestBuilder body(String body) {
        if (nullOrEmpty(body))
            return this;
        headers.put(CONTENT_LENGTH_HEADER, String.valueOf(body.length()));
        headers.put(CONTENT_TYPE_HEADER, "application/json");
        this.body = body;
        return this;
    }

    public RequestBuilder body(InputStream data, Long size) {
        if (null == data)
            return this;
        headers.put(CONTENT_LENGTH_HEADER, String.valueOf(size));
        if (!headers.containsKey(CONTENT_TYPE_HEADER)) {
            headers.put(CONTENT_TYPE_HEADER, "application/octet-stream");
        }
        this.data = data;
        this.len = size;
        return this;
    }

    public RequestBuilder verifier(OioHttpResponseVerifier verifier) {
        this.verifier = verifier;
        return this;
    }

    /**
     * Set the context associated to the current request.
     *
     * @param reqCtx
     *            the context to get deadline and request ID from
     * @return {@code this}
     */
    public RequestBuilder withRequestContext(RequestContext reqCtx) {
        this.reqCtx = reqCtx;
        if (reqCtx != null)
            headers.put(OIO_REQUEST_ID_HEADER, reqCtx.requestId());
        return this;
    }

    /**
     * @param hosts
     *            An optional list of hosts to connect to.
     * @return this
     */
    public RequestBuilder hosts(List<InetSocketAddress> hosts) {
        this.hosts = hosts;
        return this;
    }

    public OioHttpResponse execute() throws OioException {
        if (this.hosts == null || this.hosts.isEmpty()) {
            return execute(new InetSocketAddress(uri.getHost(), uri.getPort()));
        } else {
            OioException lastExc = null;
            // TODO: implement better fallback mechanism, with randomization
            for (InetSocketAddress addr : this.hosts) {
                try {
                    if (lastExc != null)
                        logger.info("Retrying on " + addr.toString());
                    return execute(addr);
                } catch (OioException oioe) {
                    // Retry only if the cause is network
                    if (oioe.getCause() instanceof IOException) {
                        logger.warn("Failed to perform request", oioe);
                        lastExc = oioe;
                    } else {
                        throw oioe;
                    }
                }
            }
            throw new OioSystemException("HTTP request execution error", lastExc);
        }
    }

    /**
     * If a deadline has been set on the request:
     * - check it;
     * - set the socket timeout accordingly;
     * Whether a deadline has been set or not, add a timeout header
     * so the server will limit itself (99% of the timeout set on the socket).
     */
    private void applyDeadline(Socket sock) throws SocketException {
        int timeout;
        if (this.reqCtx != null) {
            if (this.reqCtx.hasDeadline())
                DeadlineManager.instance().checkDeadline(this.reqCtx.deadline());
            if (this.reqCtx.hasDeadline() || this.reqCtx.hasTimeout()) {
                timeout = this.reqCtx.timeout();
                sock.setSoTimeout(timeout);
            } else {
                // Do not change the timeout, but still set the header.
                timeout = sock.getSoTimeout();
            }
        } else {
            timeout = sock.getSoTimeout();
        }
        headers.put(OIO_TIMEOUT_HEADER, timeoutMillisToStringMicros(timeout));
    }

    private OioHttpResponse execute(InetSocketAddress addr) throws OioException {
        Socket sock = null;
        try {
            sock = socketProvider.getSocket(addr);
            applyDeadline(sock);
            if (chunked)
                sendRequestChunked(sock);
            else
                sendRequest(sock);
            OioHttpResponse resp = readResponse(sock);
            try {
                if (null != verifier)
                    verifier.verify(resp);
                return resp;
            } catch (OioException e) {
                resp.close(true);
                throw e;
            }
        } catch (IOException e) {
            if (null != sock)
                try {
                    try {
                        sock.shutdownInput();
                    } catch (SocketException se) {
                        logger.debug("Socket input already shutdown");
                    }
                    sock.close();
                } catch (IOException ioe) {
                    logger.warn("Failed to close socket, possible leak", ioe);
                }
            throw new OioSystemException("HTTP request execution error", e);
        }
    }

    public <T> T execute(Class<T> c) {
        OioHttpResponse resp = execute();
        boolean success = false;
        try {
            T t = gson().fromJson(
                    new JsonReader(new InputStreamReader(resp.body(), OIO_CHARSET)), c);
            success = true;
            return t;
        } finally {
            resp.close(success);
        }
    }

    private OioHttpResponse readResponse(Socket sock) throws IOException {
        return OioHttpResponse.build(sock, this.reqCtx);
    }

    private void sendRequest(Socket sock) throws IOException {
        headers.put("Host", uri.getHost() + ":" + uri.getPort());
        headers.put("Connection", socketProvider.reusableSocket() ? "keep-alive" : "close");
        headers.put("Accept", "*/*");
        headers.put("Accept-Encoding", "gzip, deflate");
        headers.put("User-Agent", settings.userAgent());

        if (!headers.containsKey("Content-Length"))
            headers.put(CONTENT_LENGTH_HEADER, "0");

        BufferedOutputStream bos = new BufferedOutputStream(sock.getOutputStream(),
                settings.sendBufferSize());
        bos.write(requestHead());
        if (null != data) {
            stream(bos);
        } else if (null != body) {
            bos.write(body.getBytes(OIO_CHARSET));
        }
        bos.flush();
    }

    private void sendRequestChunked(Socket sock) throws IOException {
        headers.put("Host", uri.getHost() + ":" + uri.getPort());
        headers.put("Connection", socketProvider.reusableSocket() ? "keep-alive" : "close");
        headers.put("Accept", "*/*");
        headers.put("Accept-Encoding", "gzip, deflate");
        headers.put("User-Agent", settings.userAgent());

        // ensure no content-length
        headers.remove("Content-Length");
        BufferedOutputStream bos = new BufferedOutputStream(sock.getOutputStream(),
                settings.sendBufferSize());
        byte[] requestHead = requestHead();
        bos.write(requestHead);
        streamChunked(bos);
        bos.flush();
    }

    private void streamChunked(OutputStream os) throws IOException {
        byte[] b = new byte[settings.sendBufferSize()];
        int remaining = len.intValue();
        while (remaining > 0) {
            int read = data.read(b, 0, Math.min(remaining, b.length));
            if (-1 == read)
                throw new EOFException("Unexpected end of source stream");
            remaining -= read;
            if (read > 0) {
                os.write((Integer.toHexString(read) + CRLF).getBytes(OIO_CHARSET));
                os.write(b, 0, read);
                os.write(CRLF_BYTES);
            }
        }
        os.write(("0" + CRLF + CRLF).getBytes(OIO_CHARSET));
    }

    private void stream(OutputStream sos) throws IOException {
        byte[] b = new byte[settings.sendBufferSize()];
        int remaining = len.intValue();

        while (remaining > 0) {
            int read = data.read(b, 0, Math.min(remaining, b.length));
            if (-1 == read)
                throw new EOFException("Unexpected end of source stream");
            remaining -= read;
            sos.write(b, 0, read);
        }
    }

    private byte[] requestHead() throws UnsupportedEncodingException {
        StringBuilder qbuilder = new StringBuilder(null == uri.getRawQuery() ? ""
                : uri.getRawQuery());
        boolean removeTrailindAnd = 0 == qbuilder.length();
        for (Entry<String, String> h : query.entrySet()) {
            qbuilder.append("&").append(URLEncoder.encode(h.getKey(), "utf-8")).append("=")
                    .append(URLEncoder.encode(h.getValue(), "utf-8"));
        }
        String uriPath = uri.getRawPath();
        StringBuilder req = new StringBuilder(method)
                .append(" ")
                .append(uriPath != null && !uriPath.isEmpty() ? uriPath : "/")
                .append(qbuilder.length() > 0 ? "?" : "")
                .append(qbuilder.length() > 0 ? (removeTrailindAnd ? qbuilder.substring(1)
                        : qbuilder.toString()) : "").append(" ").append("HTTP/1.1")
                .append(CRLF);
        for (Entry<String, String> h : headers.entrySet()) {
            req.append(h.getKey()).append(": ").append(h.getValue()).append(CRLF);
        }
        byte[] res = req.append(CRLF).toString().getBytes(OIO_CHARSET);
        return res;
    }
} public interface SdsLogger {

    /**
     * Logs a message at {@link Level#FINEST}.
     *
     * @param message
     *            the message to log.
     */
    public void trace(String message);

    /**
     * Logs a throwable at {@link Level#FINEST}. The message of the Throwable
     * will be the message.
     *
     * @param thrown
     *            the Throwable to log.
     */
    public void trace(Throwable thrown);

    /**
     * Logs message with associated throwable information at
     * {@link Level#FINEST}.
     *
     * @param message
     *            the message to log
     * @param thrown
     *            the Throwable associated to the message.
     */
    public void trace(String message, Throwable thrown);

    /**
     * Checks if the {@link Level#FINEST} is enabled.
     *
     * @return true if enabled, false otherwise.
     */
    public boolean isTraceEnabled();

    /**
     * Logs a message at {@link Level#FINE}.
     *
     * @param message
     *            the message to log.
     */
    public void debug(String message);
    
    /**
     * Logs message with associated throwable information at
     * {@link Level#FINE}.
     *
     * @param message
     *            the message to log
     * @param thrown
     *            the Throwable associated to the message.
     */
    public void debug(String message, Throwable thrown);

    /**
     * Checks if the {@link Level#FINE} is enabled.
     *
     * @return true if enabled, false otherwise.
     */
    public boolean isDebugEnabled();

    /**
     * Logs a message at {@link Level#INFO}.
     *
     * @param message
     *            the message to log.
     */
    public void info(String message);

    /**
     * Logs a message at {@link Level#WARNING}.
     *
     * @param message
     *            the message to log.
     */
    public void warn(String message);

    /**
     * Logs a throwable at {@link Level#WARNING}. The message of the Throwable
     * will be the message.
     *
     * @param thrown
     *            the Throwable to log.
     */
    public void warn(Throwable thrown);

    /**
     * Logs message with associated throwable information at
     * {@link Level#WARNING}.
     *
     * @param message
     *            the message to log
     * @param thrown
     *            the Throwable associated to the message.
     */
    public void warn(String message, Throwable thrown);

    /**
     * Logs a message at {@link Level#SEVERE}.
     *
     * @param message
     *            the message to log.
     */
    public void error(String message);

    /**
     * Logs a throwable at {@link Level#SEVERE}. The message of the Throwable
     * will be the message.
     *
     * @param thrown
     *            the Throwable to log.
     */
    public void error(Throwable thrown);

    /**
     * Logs message with associated throwable information at
     * {@link Level#SEVERE}.
     *
     * @param message
     *            the message to log
     * @param thrown
     *            the Throwable associated to the message.
     */
    public void error(String message, Throwable thrown);
} public class SdsLoggerFactory {

    private static int loaded = -1;

    public static SdsLogger getLogger(Class<?> c) {
        ensureLoaded();
        switch (loaded) {
        case 1:
            return new Log4jLogger(c);
        case 0:
        default:
            return new BaseLogger(c);
        }
    }

    private static void ensureLoaded() {
        if (-1 == loaded) {
            try {
                Class.forName("org.apache.log4j.LogManager");
                loaded = 1;
            } catch (ClassNotFoundException e) {
                loaded = 0;
            }
        }
    }

} public class ChunkInfo {

    public ChunkInfo() {
    }

    private String url;
    private String real_url;
    private Long size;
    private String hash;
    private Position pos;

    public String url() {
        return url;
    }

    public String hash() {
        return hash;
    }

    public Position pos() {
        return pos;
    }

    public Long size() {
        return size;
    }

    public ChunkInfo url(String url) {
        this.url = url;
        return this;
    }

    public ChunkInfo real_url(String real_url) {
        this.real_url = real_url;
        return this;
    }

    public ChunkInfo size(Long size) {
        this.size = size;
        return this;
    }

    public ChunkInfo hash(String hash) {
        this.hash = hash;
        return this;
    }

    public ChunkInfo pos(Position pos) {
        this.pos = pos;
        return this;
    }

    public String id(){
        return url.substring(url.lastIndexOf("/") + 1);
    }

    public String finalUrl() {
        if (real_url != null && real_url.length() > 0) {
            return real_url;
        }
        return url;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .omitNullValues()
                .add("url", url)
                .add("real_url", real_url)
                .add("size", size)
                .add("hash", hash)
                .add("pos", pos)
                .toString();
    }
} public class ObjectInfo {

    private OioUrl url;
    private String oid;
    private Long ctime;
    private Boolean deleted = false;
    private String policy;
    private String hash;
    private String hashMethod;
    private String chunkMethod;
    private Long size;
    private Long version;
    private String mimeType;
    private Map<String, String> properties;
    private List<ChunkInfo> chunks;
    private ECInfo ecinfo;
    private RequestContext reqCtx;

    private transient Map<Integer, List<ChunkInfo>> sortedChunks;

    private static final Comparator<ChunkInfo> comparator = new Comparator<ChunkInfo>() {

        @Override
        public int compare(ChunkInfo c1, ChunkInfo c2) {
            return c1.pos()
                    .compare(c2.pos());
        }
    };

    public ObjectInfo() {
        // TODO: content hash
        this.properties = new HashMap<String, String>();
        this.hash = Hash.md5()
                .hashBytes("".getBytes(OIO_CHARSET))
                .toString();
    }

    public List<ChunkInfo> chunks() {
        return chunks;
    }

    public Map<Integer, List<ChunkInfo>> sortedChunks() {
        return sortedChunks;
    }

    public ObjectInfo chunks(List<ChunkInfo> chunks) {
        this.sortedChunks = sortChunks(chunks);
        this.chunks = chunks;
        return this;
    }

    public ObjectInfo size(Long size) {
        this.size = size;
        return this;
    }

    public ObjectInfo hash(String hash) {
        this.hash = hash;
        return this;
    }

    public ObjectInfo url(OioUrl url) {
        this.url = url;
        return this;
    }

    public OioUrl url() {
        return url;
    }

    public String oid() {
        return oid;
    }

    public ObjectInfo oid(String oid) {
        this.oid = oid;
        return this;
    }

    public String hash() {
        return hash;
    }

    public Long size() {
        return size;
    }

    public String policy() {
        return policy;
    }

    public ObjectInfo policy(String policy) {
        this.policy = policy;
        return this;
    }

    public boolean deleted() {
        return deleted;
    }

    public Long ctime() {
        return ctime;
    }

    public ObjectInfo ctime(Long ctime) {
        this.ctime = ctime;
        return this;
    }

    public String hashMethod() {
        return hashMethod;
    }

    public ObjectInfo hashMethod(String hashMethod) {
        this.hashMethod = hashMethod;
        return this;
    }

    public String chunkMethod() {
        return chunkMethod;
    }

    public ObjectInfo chunkMethod(String chunkMethod) {
        this.chunkMethod = chunkMethod;
        this.ecinfo = ECInfo.fromString(chunkMethod);
        return this;
    }

    public Long version() {
        return version;
    }

    public ObjectInfo version(Long version) {
        this.version = version;
        return this;
    }

    @Deprecated
    public String mtype() {
        return mimeType();
    }

    @Deprecated
    public ObjectInfo mtype(String mimeType) {
        return mimeType(mimeType);
    }

    public String mimeType() {
        return mimeType;
    }

    public ObjectInfo mimeType(String mimeType) {
        this.mimeType = mimeType;
        return this;
    }

    public Integer nbchunks() {
        return isEC() ? chunks.size() : sortedChunks.size();
    }

    public Long chunksize(Integer pos) {
        return sortedChunks.get(pos).get(0).size();
    }

    public Map<String, String> properties() {
        return properties;
    }

    public ObjectInfo properties(Map<String, String> properties) {
        this.properties = properties;
        return this;
    }

    public ECInfo ecinfo(){
    	return ecinfo;
    }

    public boolean isEC(){
    	return null != ecinfo;
    }

    public long metachunksize(int pos) {
        if (!isEC())
            return -1;
        long maxmcsize = ecinfo.k() * sortedChunks.get(pos).get(0).size();
        long remaining = size - (pos * maxmcsize);
        return Math.min(maxmcsize, remaining);
    }

    /**
     * @return a string suitable to be sent as {@code CHUNK_META_FULL_PATH} header value.
     */
    public String fullpath() {
        if (this.url.account() == null || this.url.account().isEmpty()
                || this.url.container() == null || this.url.container().isEmpty()
                || this.url.object() == null || this.url.object().isEmpty()
                || this.version() == null || this.version() == 0
                || this.oid() == null || this.oid().isEmpty()) {
            throw new IllegalArgumentException("Can't encode fullpath");
        }
        StringBuilder sb = new StringBuilder();
        sb.append(Strings.quote(this.url.account()));
        sb.append('/');
        sb.append(Strings.quote(this.url.container()));
        sb.append('/');
        sb.append(Strings.quote(this.url.object()));
        sb.append('/');
        sb.append(Strings.quote(this.version()));
        sb.append('/');
        sb.append(Strings.quote(this.oid()));
        return sb.toString();
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .omitNullValues()
                .add("url", url)
                .add("ctime", ctime)
                .add("mime-type", mimeType)
                .add("deleted", deleted)
                .add("policy", policy)
                .add("hash", hash)
                .add("hash-method", hashMethod)
                .add("chunk-method", chunkMethod)
                .add("size", size)
                .add("version", version)
                .add("props", properties)
                .add("chunks", chunks)
                .toString();
    }

    public RequestContext requestContext() {
        return this.reqCtx;
    }

    /**
     * Set the context of the request that lead to this object's creation.
     *
     * @param reqCtx
     *            the request context
     * @return {@code this}
     */
    public ObjectInfo withRequestContext(RequestContext reqCtx) {
        this.reqCtx = reqCtx;
        return this;
    }

    /* -- INTERNAL -- */

    private Map<Integer, List<ChunkInfo>> sortChunks(
            List<ChunkInfo> chunks) {

        Map<Integer, List<ChunkInfo>> res = new HashMap<Integer, List<ChunkInfo>>();
        for (ChunkInfo ci : chunks) {
            List<ChunkInfo> l = res.get(ci.pos().meta());
            if (null == l) {
                l = new ArrayList<ChunkInfo>();
                res.put(ci.pos().meta(), l);
            }
            l.add(ci);
        }
        for (List<ChunkInfo> l : res.values())
            Collections.sort(l, comparator);

        return res;
    }

} public class Range {

    private static final Pattern RANGE_PATTERN = Pattern
            .compile("^([\\d]+)?-([\\d]+)?$");

    private long from = 0;
    private long to = -1;

    private Range(long from, long to) {
        this.from = from;
        this.to = to;
    }

    public static Range upTo(long to) {
        checkArgument(0 < to);
        return new Range(0, to);
    }

    public static Range from(long from) {
        checkArgument(0 <= from);
        return new Range(from, -1);
    }

    public static Range between(long from, long to) {
        checkArgument(from >= 0 && to > 0 && to >= from,
                "Invalid range");
        return new Range(from, to);
    }

    public static Range parse(String str) {
        Matcher m = RANGE_PATTERN.matcher(str);
        checkArgument(m.matches());
        if (null == m.group(1)) {
            checkArgument(null != m.group(2), "useless range");
            return upTo(parseInt(m.group(2)));
        }
        return (null == m.group(2)) ? from(parseInt(m.group(1)))
                : between(parseInt(m.group(1)), parseInt(m.group(2)));
    }

    public long from() {
        return from;
    }

    public long to() {
        return to;
    }

    public String headerValue() {
        return to < 0
                ? format("bytes=%d-", from)
                : format("bytes=%d-%d", from, to);
    }

    public String rangeValue() {
        return to < 0
                ? format("%d-", from)
                : format("%d-%d", from, to);
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("from", from)
                .add("to", to)
                .toString();
    }
} public class DownloadHelper {

	public static List<Target> loadTargets(ObjectInfo oinf, Range range) {

		LinkedList<Target> targets = new LinkedList<Target>();

		if (null == range) {
			for (int i = 0; i < oinf.sortedChunks().size(); i++) {
				targets.addFirst(new Target()
				        .setChunk(oinf.sortedChunks().get(i)));
			}
		} else {
			ObjectOffset begin = findOffset(oinf, range.from());
			ObjectOffset end = findOffset(oinf, range.to());
			if (end.pos() == begin.pos()) {
				targets.addFirst(new Target()
				        .setChunk(oinf.sortedChunks().get(begin.pos()))
				        .setRange(between(begin.offset(), end.offset())));
			} else {
				targets.addFirst(new Target()
				        .setChunk(oinf.sortedChunks().get(begin.pos()))
				        .setRange(between(begin.offset(),
				                oinf.chunksize(begin.pos()).intValue()
				                        - 1)));
				for (int extrapos = begin.pos() + 1; extrapos < end
				        .pos(); extrapos++) {
					targets.addFirst(new Target()
					        .setChunk(oinf.sortedChunks().get(extrapos))
					        .setRange(between(0,
					                oinf.chunksize(extrapos).intValue())));
				}

				targets.addFirst(new Target()
				        .setChunk(oinf.sortedChunks().get(end.pos()))
				        .setRange(between(0, end.offset())));
			}
		}
		Collections.reverse(targets);
		return targets;
	}

	private static ObjectOffset findOffset(ObjectInfo oinf, long offset) {
		if (-1 == offset)
			return new ObjectOffset()
			        .pos(oinf.nbchunks() - 1)
			        .offset(oinf.chunksize(oinf.nbchunks() - 1).intValue() - 1);
		for (int pos = 0; pos < oinf.nbchunks(); pos++) {
			long size = oinf.chunksize(pos);
			if (offset <= size)
				return new ObjectOffset().pos(pos).offset(offset);
			offset -= size;
		}
		throw new IllegalArgumentException("Range begin out of content size");
	}
} public interface StorageClient {

    public ObjectInfo uploadChunks(ObjectInfo oinf, InputStream data);

    public ObjectInfo uploadChunks(ObjectInfo oinf, InputStream data, RequestContext reqCtx);

    public ObjectInfo uploadChunks(ObjectInfo oinf, File data);

    public ObjectInfo uploadChunks(ObjectInfo oinf, File data, RequestContext reqCtx);

    public ObjectInfo uploadChunks(ObjectInfo oinf, byte[] data);

    public ObjectInfo uploadChunks(ObjectInfo oinf, byte[] data, RequestContext reqCtx);

    public InputStream downloadObject(ObjectInfo oinf);

    public InputStream downloadObject(ObjectInfo oinf, Range range);

    public InputStream downloadObject(ObjectInfo oinf, RequestContext reqCtx);

    public InputStream downloadObject(ObjectInfo oinf, Range range, RequestContext reqCtx);

} public class Target {

	private List<ChunkInfo> ci;
	private Range range;
	
	public Target() {
	}

	public List<ChunkInfo> getChunk() {
		return ci;
	}

	public Target setChunk(List<ChunkInfo> ci) {
		this.ci = ci;
		return this;
	}

	public Range getRange() {
		return range;
	}

	public Target setRange(Range range) {
		this.range = range;
		return this;
	}
	
	
	
} public static void checkArgument(boolean condition, String msg) {
    if (!condition)
        throw new IllegalArgumentException(msg);
} public static final String CHUNK_META_CHUNK_HASH = "X-oio-chunk-meta-chunk-hash"; public static final String CHUNK_META_CHUNK_ID = "X-oio-chunk-meta-chunk-id"; public static final String CHUNK_META_CHUNK_POS = "X-oio-chunk-meta-chunk-pos"; public static final String CHUNK_META_CONTAINER_ID = "X-oio-chunk-meta-container-id"; public static final String CHUNK_META_CONTENT_CHUNKSNB = "x-oio-chunk-meta-content-chunksnb"; public static final String CHUNK_META_CONTENT_CHUNK_METHOD = "X-oio-chunk-meta-content-chunk-method"; public static final String CHUNK_META_CONTENT_ID = "X-oio-chunk-meta-content-id"; public static final String CHUNK_META_CONTENT_MIME_TYPE = "X-oio-chunk-meta-content-mime-type"; public static final String CHUNK_META_CONTENT_PATH = "X-oio-chunk-meta-content-path"; public static final String CHUNK_META_CONTENT_POLICY = "X-oio-chunk-meta-content-storage-policy"; public static final String CHUNK_META_CONTENT_SIZE = "X-oio-chunk-meta-content-size"; public static final String CHUNK_META_CONTENT_VERSION = "X-oio-chunk-meta-content-version"; public static final String CHUNK_META_FULL_PATH = "X-oio-chunk-meta-full-path"; public static final String CHUNK_META_OIO_VERSION = "X-oio-chunk-meta-oio-version"; public static final OioHttpResponseVerifier RAWX_VERIFIER = new OioHttpResponseVerifier() {

    @Override
    public void verify(OioHttpResponse resp) throws OioException {
        switch (resp.code()) {
        case 200:
        case 201:
        case 204:
        case 206:
            return;
        case 400:
            char[] buf = new char[256];
            try {
                new InputStreamReader(resp.body()).read(buf);
                throw new BadRequestException(resp.msg() + new String(buf));
            } catch (IOException e) {
                e.printStackTrace();
                throw new BadRequestException(resp.msg());
            }
        case 404:
            throw new ChunkNotFoundException(resp.msg());
        case 500:
            throw new OioException(format("Internal error (%d %s)",
                    resp.code(), resp.msg()));
        default:
            throw new OioException(format("Unmanaged response code (%d %s)",
                    resp.code(), resp.msg()));
        }
    }
};
import io.openio.sds.RequestContext;
import io.openio.sds.common.FeedableInputStream;
import io.openio.sds.common.Hex;
import io.openio.sds.common.Strings;
import io.openio.sds.exceptions.OioException;
import io.openio.sds.http.OioHttp;
import io.openio.sds.http.OioHttp.RequestBuilder;
import io.openio.sds.logging.SdsLogger;
import io.openio.sds.logging.SdsLoggerFactory;
import io.openio.sds.models.ChunkInfo;
import io.openio.sds.models.ObjectInfo;
import io.openio.sds.models.Range;
import io.openio.sds.storage.DownloadHelper;
import io.openio.sds.storage.StorageClient;
import io.openio.sds.storage.Target;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import static io.openio.sds.common.Check.checkArgument;
import static io.openio.sds.common.OioConstants.CHUNK_META_CHUNK_HASH;
import static io.openio.sds.common.OioConstants.CHUNK_META_CHUNK_ID;
import static io.openio.sds.common.OioConstants.CHUNK_META_CHUNK_POS;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTAINER_ID;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_CHUNKSNB;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_CHUNK_METHOD;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_ID;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_MIME_TYPE;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_PATH;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_POLICY;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_SIZE;
import static io.openio.sds.common.OioConstants.CHUNK_META_CONTENT_VERSION;
import static io.openio.sds.common.OioConstants.CHUNK_META_FULL_PATH;
import static io.openio.sds.common.OioConstants.CHUNK_META_OIO_VERSION;
import static io.openio.sds.http.Verifiers.RAWX_VERIFIER;
import static java.lang.String.format;
import static java.nio.ByteBuffer.wrap;
package io.openio.sds.storage.rawx;




/**
 * @author Christopher Dedeurwaerder
 */
public class RawxClient implements StorageClient {

	private static final SdsLogger logger = SdsLoggerFactory
			.getLogger(RawxClient.class);

	private static final int MIN_WORKERS = 1;
	private static final int MAX_WORKERS = 100;
	private static final int IDLE_THREAD_KEEP_ALIVE = 30; // in seconds

	final OioHttp http;
	private final ExecutorService executors;
	private final RawxSettings settings;

	public RawxClient(OioHttp http, RawxSettings settings) {
		this.http = http;
		this.settings = settings;
		this.executors = new ThreadPoolExecutor(MIN_WORKERS,
				MAX_WORKERS,
				IDLE_THREAD_KEEP_ALIVE,
				TimeUnit.SECONDS,
				new SynchronousQueue<Runnable>(),
				new ThreadFactory() {

					@Override
					public Thread newThread(Runnable r) {
						Thread t = new Thread(r);
						t.setName("RawxClient-Worker");
						return t;
					}
				});
	}

	public static RawxClient client(OioHttp http,
									RawxSettings settings) {
		checkArgument(null != http, "AsynHttpClient cannot be null");
		checkArgument(null != settings, "Settings cannot be null");
		return new RawxClient(http, settings);
	}

	public int getActiveUploadCount() {
		return ((ThreadPoolExecutor) this.executors).getActiveCount();
	}

	/**
	 * Uploads the chunks of the specified {@code ObjectInfo} asynchronously
	 *
	 * @param oinf the ObjectInfo to deal with
	 * @param data the data to upload
	 * @return {@code oinf}
	 */
	public ObjectInfo uploadChunks(ObjectInfo oinf,
								   InputStream data) {
		return uploadChunks(oinf, data, new RequestContext());
	}

	/**
	 * Uploads the chunks of the specified {@code ObjectInfo} asynchronously
	 *
	 * @param oinf  the ObjectInfo to deal with
	 * @param data  the data to upload
	 * @param reqCtx Common parameters to all requests
	 * @return {@code oinf}
	 */
	public ObjectInfo uploadChunks(ObjectInfo oinf,
								   InputStream data, RequestContext reqCtx) {
		StreamWrapper wrapper = new StreamWrapper(data);
		long remaining = oinf.size();
		for (int pos = 0; pos < oinf.nbchunks(); pos++) {
			long csize = Math.min(remaining, oinf.chunksize(pos));
			if (csize == 0 && pos != 0)
				throw new OioException("Too many chunks prepared");
			uploadPosition(oinf, pos, csize, wrapper, reqCtx);
			remaining -= csize;
		}
		return oinf.hash(Hex.toHex(wrapper.md5()));
	}

	/**
	 * Uploads the chunks of the specified {@code ObjectInfo} asynchronously
	 *
	 * @param oinf the ObjectInfo to deal with
	 * @param data the data to upload
	 * @return oinf
	 */
	public ObjectInfo uploadChunks(ObjectInfo oinf, File data) {
		return uploadChunks(oinf, data, new RequestContext());
	}

	/**
	 * Uploads the chunks of the specified {@code ObjectInfo} asynchronously
	 *
	 * @param oinf  the ObjectInfo to deal with
	 * @param data  the data to upload
	 * @param reqCtx Common parameters to all requests
	 * @return oinf
	 */
	public ObjectInfo uploadChunks(ObjectInfo oinf, File data, RequestContext reqCtx) {

		try {
			FileInputStream fin = new FileInputStream(data);
			try {
				return uploadChunks(oinf, fin, reqCtx);
			} finally {
				try {
					fin.close();
				} catch (IOException e) {
					logger.warn("Failed to close Inputstream, possible leak", e);
				}
			}
		} catch (FileNotFoundException e) {
			throw new IllegalArgumentException("File not found", e);
		}
	}

	public ObjectInfo uploadChunks(ObjectInfo oinf, byte[] data) {
		return uploadChunks(oinf, data, new RequestContext());
	}

	public ObjectInfo uploadChunks(
			ObjectInfo oinf, byte[] data, RequestContext reqCtx) {
		return uploadChunks(oinf, new ByteArrayInputStream(data), reqCtx);
	}

	public InputStream downloadObject(ObjectInfo oinf) {
		return downloadObject(oinf, new RequestContext());
	}

	public InputStream downloadObject(ObjectInfo oinf, Range range) {
		return downloadObject(oinf, range, new RequestContext());
	}

	public InputStream downloadObject(ObjectInfo oinf, RequestContext reqCtx) {
		return downloadObject(oinf, null, reqCtx);
	}

	public InputStream downloadObject(ObjectInfo oinf, Range range, RequestContext reqCtx) {
		checkArgument(null != oinf);
		List<Target> targets = DownloadHelper.loadTargets(oinf, range);
		return new ObjectInputStream(targets, http, reqCtx);
	}

	public void deleteChunks(List<ChunkInfo> l) {
		for (ChunkInfo ci : l)
			deleteChunk(ci);
	}

	public void deleteChunk(ChunkInfo ci) {
		// no verifier, suppress exceptions
		try {
			http.delete(ci.finalUrl())
					.execute()
					.close();
		} catch (OioException e) {
			if (logger.isDebugEnabled())
				logger.debug(format("Chunk %s deletion error", ci.url()), e);
		}
	}

	/* --- INTERNALS --- */

	private <E> void quorumOrFail(int pos, int quorum, List<E> successes) {
		if (successes.size() < quorum) {
			throw new OioException(format(
					"Quorum not reached when writing chunks at position %s (%d/%d)",
					pos, successes.size(), quorum));
		}
	}

	private void cancelTasks(List<Future<UploadResult>> futures, Exception exc,
							 String prefixMessage) {
		int notTerminated = 0;
		for (Future<UploadResult> future : futures) {
			if (!future.isDone())
				notTerminated++;
			future.cancel(true);
		}
		String message = prefixMessage;
		if (notTerminated > 0)
			message += " (" + notTerminated + " upload jobs cancelled)";
		throw new OioException(message, exc);
	}

	private ObjectInfo uploadPosition(final ObjectInfo oinf, final int pos, final Long size,
									  InputStream data, final RequestContext reqCtx) {
		List<ChunkInfo> cil = oinf.sortedChunks().get(pos);
		final List<FeedableInputStream> gens = size == 0 ? null : feedableBodies(cil.size(), size);
		List<Future<UploadResult>> futures = new ArrayList<Future<UploadResult>>();

		int quorum;
		if (!settings.quorumWrite())
			quorum = cil.size();
		else
			quorum = (cil.size() + 1) / 2;

		for (int i = 0; i < cil.size(); i++) {
			final ChunkInfo ci = cil.get(i);
			final FeedableInputStream in = null == gens ? null : gens.get(i);

			Callable<UploadResult> uploader = new Callable<UploadResult>() {

				@Override
				public UploadResult call() {
					UploadResult result = new UploadResult(ci);
					try {
						RequestBuilder builder = http
								.put(ci.finalUrl())
								.header(CHUNK_META_CONTAINER_ID, oinf.url().cid())
								.header(CHUNK_META_CONTENT_ID, oinf.oid())
								.header(CHUNK_META_CONTENT_VERSION, String.valueOf(oinf.version()))
								.header(CHUNK_META_CONTENT_POLICY, oinf.policy())
								.header(CHUNK_META_CONTENT_MIME_TYPE, oinf.mimeType())
								.header(CHUNK_META_CONTENT_CHUNK_METHOD, oinf.chunkMethod())