/*
 * Copyright 2016-2017 Leon Chen
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.moilioncircle.redis.rdb.cli.net.protocol;

import static com.moilioncircle.redis.replicator.Constants.COLON;
import static com.moilioncircle.redis.replicator.Constants.DOLLAR;
import static com.moilioncircle.redis.replicator.Constants.MINUS;
import static com.moilioncircle.redis.replicator.Constants.PLUS;
import static com.moilioncircle.redis.replicator.Constants.STAR;

import java.io.IOException;
import java.io.OutputStream;

import com.moilioncircle.redis.rdb.cli.util.ByteBuffers;
import com.moilioncircle.redis.replicator.io.RedisInputStream;
import com.moilioncircle.redis.replicator.util.ByteBuilder;

/**
 * @author Baoyi Chen
 */
public class Protocol {
    
    private final RedisInputStream in;
    private final OutputStream out;
    
    public Protocol(RedisInputStream in, OutputStream out) {
        this.in = in;
        this.out = out;
    }
    
    public void emit(byte[] command, byte[]... ary) throws IOException {
        Protocols.emit(out, command, ary);
    }
    
    public void emit(ByteBuffers command, ByteBuffers... ary) throws IOException {
        Protocols.emit(out, command, ary);
    }

    public RedisObject parse() throws IOException {
        while (true) {
            int c = in.read();
            switch (c) {
                case DOLLAR:
                    // RESP Bulk Strings
                    ByteBuilder builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    long len = Long.parseLong(builder.toString());
                    if (len == -1) return new RedisObject(RedisObject.Type.NULL, null);
                    RedisObject r = new RedisObject(RedisObject.Type.BULK, in.readBytes(len).first());
                    if ((c = in.read()) != '\r') throw new RuntimeException("expect '\\r' but :" + (char) c);
                    if ((c = in.read()) != '\n') throw new RuntimeException("expect '\\n' but :" + (char) c);
                    return r;
                case COLON:
                    // RESP Integers
                    builder = ByteBuilder.allocate(128);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    // As integer
                    return new RedisObject(RedisObject.Type.NUMBER, Long.parseLong(builder.toString()));
                case STAR:
                    // RESP Arrays
                    builder = ByteBuilder.allocate(128);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            break;
                        } else {
                            builder.put((byte) c);
                        }
                    }
                    len = Long.parseLong(builder.toString());
                    if (len == -1) return new RedisObject(RedisObject.Type.NULL, null);
                    RedisObject[] ary = new RedisObject[(int) len];
                    for (int i = 0; i < len; i++) {
                        RedisObject obj = parse();
                        ary[i] = obj;
                    }
                    return new RedisObject(RedisObject.Type.ARRAY, ary);
                case PLUS:
                    // RESP Simple Strings
                    builder = ByteBuilder.allocate(128);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            return new RedisObject(RedisObject.Type.STRING, builder.array());
                        } else {
                            builder.put((byte) c);
                        }
                    }
                case MINUS:
                    // RESP Errors
                    builder = ByteBuilder.allocate(32);
                    while (true) {
                        while ((c = in.read()) != '\r') {
                            builder.put((byte) c);
                        }
                        if ((c = in.read()) == '\n') {
                            return new RedisObject(RedisObject.Type.ERR, builder.array());
                        } else {
                            builder.put((byte) c);
                        }
                    }
                default:
                    throw new RuntimeException("expect [$,:,*,+,-] but: " + (char) c);

            }
        }
    }
}
