package meerkat.comm;

import com.google.protobuf.Message;

import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.InvocationTargetException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * Created by Arbel Deutsch Peled on 21-Feb-16.
 * A input stream of Protobuf messages
 */
public class MessageInputStream<T extends Message> implements Iterable<T>{

    Class<T> type;

    private T.Builder builder;

    private InputStream in;

    MessageInputStream(InputStream in, Class<T> type) throws IOException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        this.type = type;
        this.in = in;
        this.builder = (T.Builder) type.getMethod("newBuilder").invoke(type);
    }

    @Override
    public Iterator<T> iterator() {

        return new Iterator<T>() {

            @Override
            public boolean hasNext() {
                try {
                    return isAvailable();
                } catch (IOException e) {
                    return false;
                }
            }

            @Override
            public T next() {
                try {
                    return readMessage(type);
                } catch (IOException e) {
                    return null;
                }
            }

        };

    }

    /**
     * Factory class for actually creating a MessageInputStream
     */
    public static class MessageInputStreamFactory {

        public static <T extends Message> MessageInputStream<T> createMessageInputStream(InputStream in, Class<T> type)
                throws IOException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {

            return new MessageInputStream<>(in, type);

        }

    }

    public T readMessage(Class<T> msgType) throws IOException{
        builder.clear();
        builder.mergeDelimitedFrom(in);
        return msgType.cast(builder.build());
    }

    public boolean isAvailable() throws IOException {
        return (in.available() > 0);
    }

    public List<T> asList() throws IOException{

        List<T> list = new LinkedList<>();

        while (isAvailable()){
            list.add(readMessage(type));
        }

        return list;

    }

    public void close() throws IOException {

        in.close();

    }

}
