package org.freshvanilla.rmi;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.freshvanilla.net.CachedDataSocketFactory;
import org.freshvanilla.net.DataSocket;
import org.freshvanilla.net.WireFormat;
import org.freshvanilla.utils.*;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLStreamHandler;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.*;
import static java.util.Arrays.asList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class Proxies {
    private static final Log LOG = LogFactory.getLog(Proxies.class);
    private static final NumberFormat DECIMAL = new DecimalFormat();
    private static final Executor logger = Executors.newSingleThreadExecutor(new NamedThreadFactory("logger", Thread.MIN_PRIORITY, true));
    public static final String NODE_ID = "node-" + Long.toString(System.currentTimeMillis(), 36);

    private Proxies() {
    }

    public static <P> VanillaRmiServer<P> newServer(String name, int port, P provider) throws IOException {
        return new VanillaRmiServer<P>(name, port, provider);
    }

    public static <I> I newClient(Factory<String, DataSocket> factory, Class<I>... interfaces) {
        return newClient(factory, false, interfaces);
    }

    public static <I> I newClient(String name, String connectionString, Class<I>... interfaces) {
        final Factory<String, DataSocket> factory = new CachedDataSocketFactory(name, connectionString, 60 * 1000L);
        return newClient(factory, true, interfaces);
    }

    public static <I> I newClient(Factory<String, DataSocket> factory, boolean closeFactory, Class<I>... interfaces) {
        Set<Class> interfaceSet = new LinkedHashSet<Class>(Arrays.asList(interfaces));
        interfaceSet.add(SimpleResource.class);
        final Class[] interfaces2 = interfaceSet.toArray(new Class[interfaceSet.size()]);

        final RmiInvocationHandler rmiih = new RmiInvocationHandler(factory, closeFactory);
        return (I) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), interfaces2, rmiih);
    }

    static class RmiInvocationHandler implements InvocationHandler {
        private static final Object[] NO_OBJECTS = {};
        private final Factory<String, DataSocket> factory;
        private final boolean closeFactory;

        RmiInvocationHandler(Factory<String, DataSocket> factory, boolean closeFactory) {
            this.factory = factory;
            this.closeFactory = closeFactory;
        }

        public Object invoke(Object proxy, Method method, Object... args) throws Throwable {
            if (args == null || args.length == 0) {
                final String methodName = method.getName();
                if ("isClosed".equals(methodName)) {
                    return factory.isClosed();
                } else if ("getName".equals(methodName)) {
                    return factory.getName();
                } else if ("close".equals(methodName)) {
                    close();
                    return null;
                }
            }
            RmiMethod rmiMethod = getRmiMethod(method);
            final Class<?>[] parameterTypes = rmiMethod.parameterTypes;
            boolean async = parameterTypes.length > 0 && parameterTypes[parameterTypes.length - 1] == Callback.class;
            if (args == null) args = NO_OBJECTS;
            int argsLength = args.length - (async ? 1 : 0);
            final DataSocket ds = factory.acquire(async ? "async-org.freshvanilla.rmi" : "sync-org.freshvanilla.rmi");
            try {
                // write the sequence number.
                final long sequenceNumber = async ? ds.microTimestamp() : 0;
                if (async) {
                    Callback callback = (Callback) args[args.length - 1];
                    ds.callbackMap().put(sequenceNumber, callback);
                    ds.setReader(new RmiCallback(ds));
                }

                final WireFormat wf = ds.wireFormat();
                ByteBuffer wb = ds.writeBuffer();
                wf.writeNum(wb, sequenceNumber);
                wf.writeTag(wb, rmiMethod.methodName);
                wf.writeArray(wb, argsLength, args);
                wf.flush(ds, wb);

                if (async)
                    return null;

                ByteBuffer rb = ds.read();
                long sequenceNumber2 = wf.readNum(rb);
                assert sequenceNumber2 == 0;
                boolean success = wf.readBoolean(rb);
                Object reply = wf.readObject(rb);
                if (success)
                    return Classes.parseAs(reply, rmiMethod.returnType);
                if (reply instanceof Throwable) {
                    Throwable t = (Throwable) reply;
                    appendStackTrace(ds, t);
                    throw t;
                }
                throw new AssertionError(reply);
            } finally {
                factory.recycle(ds);
            }
        }

        private static void appendStackTrace(DataSocket ds, Throwable t) {
            try {
                Field stackTrace = Throwable.class.getDeclaredField("stackTrace");
                stackTrace.setAccessible(true);
                final List<StackTraceElement> stack1 = asList(t.getStackTrace());
                int pos;
                for (pos = stack1.size() - 1; pos > 0; pos--)
                    if ("invoke0".equals(stack1.get(pos).getMethodName()))
                        break;
                if (pos <= 0)
                    pos = stack1.size() - 6;
                List<StackTraceElement> stack = new ArrayList(stack1.subList(0, pos));
                final InetSocketAddress address = ds.getAddress();
                if (address != null) {
                    String hostName = address.getHostName();
                    if ("0.0.0.0".equals(hostName))
                        hostName = "localhost";
                    stack.add(new StackTraceElement("~ call to server ~", "call", hostName, address.getPort()));
                }
                Throwable t2 = new Throwable();
                final List<StackTraceElement> stack2 = asList(t2.getStackTrace());
                stack.addAll(stack2.subList(3, stack2.size()));
                stackTrace.set(t, stack.toArray(new StackTraceElement[stack.size()]));
            } catch (Exception e) {
                throw new AssertionError(e);
            }
        }

        private void close() {
            if (closeFactory)
                factory.close();
        }

        private final Map<Method, RmiMethod> rmiMethodMap = new ConcurrentHashMap<Method, RmiMethod>();

        private RmiMethod getRmiMethod(Method method) {
            RmiMethod ret = rmiMethodMap.get(method);
            if (ret == null)
                rmiMethodMap.put(method, ret = new RmiMethod(method.getName(), method.getReturnType(), method.getParameterTypes()));
            return ret;
        }

        static class RmiCallback implements Callback<DataSocket> {

            private final DataSocket ds;

            RmiCallback(DataSocket ds) {
                this.ds = ds;
            }

            public void onCallback(DataSocket dataSocket) throws Exception {
                Callback<Object> callback = null;
                try {
                    ByteBuffer rb = ds.read();
                    final WireFormat wf = ds.wireFormat();
                    long sequenceNumber = wf.readNum(rb);
                    callback = ds.callbackMap().remove(sequenceNumber);
                    assert sequenceNumber != 0;
                    boolean success = wf.readBoolean(rb);
                    Object reply = wf.readObject(rb);

                    if (callback == null)
                        LOG.error("Response to unknown callback reply=" + reply);
                    else if (success)
                        callback.onCallback(reply);
                    else
                        callback.onException((Throwable) reply);
                } catch (Exception e) {
                    if (ds.isClosed()) return;
                    if (!(e instanceof AsynchronousCloseException))
                        LOG.error("Exception thrown processing callback", e);
                    try {
                        if (callback != null)
                            callback.onException(e);
                    } catch (Exception ignored) {
                        // ignored.
                    }
                }
            }

            public void onException(Throwable t) {
                LOG.warn("Unhandled exception", t);
            }
        }
    }

    static class RmiMethod {
        public final String methodName;
        public final Class returnType;
        public final Class<?>[] parameterTypes;

        RmiMethod(String methodName, Class returnType, Class<?>[] parameterTypes) {
            this.methodName = methodName;
            this.returnType = returnType;
            this.parameterTypes = parameterTypes;
        }
    }

    private static class DummyURLStreamHandler extends URLStreamHandler {
        protected URLConnection openConnection(URL u) throws IOException {
            return new URLConnection(u) {
                public void connect() throws IOException {
                    throw new IOException("connect not supported");
                }
            };
        }
    }
}
