package org.jboss.cx.remoting.core.stream;

import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedList;
import org.jboss.cx.remoting.core.util.Logger;
import org.jboss.cx.remoting.spi.protocol.MessageInput;
import org.jboss.cx.remoting.spi.protocol.MessageOutput;
import org.jboss.cx.remoting.spi.stream.RemoteStreamSerializer;
import org.jboss.cx.remoting.spi.stream.StreamContext;
import org.jboss.cx.remoting.spi.stream.StreamSerializer;
import org.jboss.cx.remoting.spi.stream.StreamSerializerFactory;

/**
 *
 */
public final class InputStreamStreamSerializerFactory implements StreamSerializerFactory {
    private static final Logger log = Logger.getLogger(InputStreamStreamSerializerFactory.class);

    public InputStreamStreamSerializerFactory() {
        // no-arg constructor required
    }

    public StreamSerializer getLocalSide(StreamContext context, Object local) throws IOException {
        return new StreamSerializerImpl(context, (InputStream)local);
    }

    public RemoteStreamSerializer getRemoteSide(final StreamContext context) throws IOException {
        return new RemoteStreamSerializerImpl(context);
    }

    private enum Type {
        DATA,
        END,
    }

    private static final int BUF_LEN = 512;

    private final static class StreamSerializerImpl implements StreamSerializer {
        private final StreamContext context;
        private final InputStream inputStream;

        public StreamSerializerImpl(final StreamContext context, final InputStream inputStream) throws IOException {
            this.context = context;
            this.inputStream = inputStream;
        }

        public void handleOpen() throws IOException {
            sendNext();
        }

        public void handleData(MessageInput data) throws IOException {
            sendNext();
        }

        private void sendNext() throws IOException {
            log.trace("Sending next message for stream %s", this);
            final MessageOutput output = context.writeMessage();
            final byte[] bytes = new byte[BUF_LEN];
            int i, t;
            boolean end = false;
            for (t = 0; t < BUF_LEN; t += i) {
                i = inputStream.read(bytes);
                if (i == -1) {
                    end = true;
                    break;
                }
            }
            if (t > 0) {
                output.write(Type.DATA.ordinal());
                output.writeInt(t);
                output.write(bytes, 0, t);
            }
            if (end) {
                output.write(Type.END.ordinal());
            }
            output.commit();
        }

        public void handleClose() throws IOException {
        }
    }

    private final static class RemoteStreamSerializerImpl implements RemoteStreamSerializer {

        private final StreamContext context;
        private LinkedList<Entry> messageQueue = new LinkedList<Entry>();

        private final class Entry {
            private final Type type;
            private final byte[] msg;
            private int i;

            public Entry(final byte[] msg) {
                this.msg = msg;
                type = Type.DATA;
            }

            public Entry() {
                type = Type.END;
                msg = null;
            }
        }

        public RemoteStreamSerializerImpl(final StreamContext context) {
            this.context = context;
        }

        public InputStream getRemoteInstance() {
            return new InputStream() {

                public int read() throws IOException {
                    boolean intr = Thread.interrupted();
                    try {
                        synchronized(messageQueue) {
                            for (;;) {
                                if (messageQueue.size() == 0) {
                                    context.writeMessage().commit();
                                    do {
                                        try {
                                            messageQueue.wait();
                                        } catch (InterruptedException e) {
                                            intr = true;
                                        }
                                    } while (messageQueue.size() == 0);
                                }
                                final RemoteStreamSerializerImpl.Entry first = messageQueue.getFirst();
                                switch (first.type) {
                                    case DATA:
                                        if (first.msg.length >= first.i) {
                                            messageQueue.removeFirst();
                                        } else {
                                            return first.msg[first.i ++];
                                        }
                                    default:
                                        return -1;
                                }
                            }
                        }
                    } finally {
                        if (intr) {
                            Thread.currentThread().interrupt();
                        }
                    }
                }
            };
        }

        public void handleOpen() throws IOException {
        }

        public void handleData(MessageInput data) throws IOException {
            synchronized(messageQueue) {
                for (;;) {
                    final int d = data.read();
                    if (d == -1) {
                        break;
                    }
                    Type t = Type.values()[d];
                    switch (t) {
                        case DATA:
                            int l = data.readInt();
                            byte[] bytes = new byte[l];
                            data.read(bytes);
                            messageQueue.add(new Entry(bytes));
                            break;
                        case END:
                            messageQueue.add(new Entry());
                            break;
                    }
                }
                messageQueue.notifyAll();
            }
        }

        public void handleClose() throws IOException {
            synchronized(messageQueue) {
                messageQueue.add(new Entry());
            }
        }
    }
}

@@ -1897,75 +1897,8 @@
 n %7B%0A
-            log.trace(%22Sending next message for stream %25s%22, this);%0A
     
@@ -1954,24 +1954,24 @@
 eMessage();%0A
+
             
@@ -2300,24 +2300,88 @@
 f (t %3E 0) %7B%0A
+                log.trace(%22Sending DATA message, %25d bytes%22, t);%0A
             
@@ -2531,24 +2531,74 @@
  if (end) %7B%0A
+                log.trace(%22Sending END message%22);%0A
             
@@ -4626,17 +4626,17 @@
 .length 
-%3E
+%3C
 = first.
@@ -4833,16 +4833,23 @@
 st.i ++%5D
+ & 0xff
 ;%0A      
@@ -5894,32 +5894,120 @@
 ta.read(bytes);%0A
+                            log.trace(%22Received DATA message; %25d bytes%22, bytes.length);%0A
                 
@@ -6081,32 +6081,32 @@
          break;%0A
-
                 
@@ -6123,16 +6123,79 @@
 se END:%0A
+                            log.trace(%22Received END message%22);%0A
         

Fix InputStreamStreamSerializer