# Pipe.java

    /*
     *  Copyright (c) 2009, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
     *
     *  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 org.apache.synapse.transport.passthru;

    import org.apache.http.TruncatedChunkException;
    import org.apache.http.nio.ContentDecoder;
    import org.apache.http.nio.ContentEncoder;
    import org.apache.http.nio.IOControl;
    import org.apache.http.nio.NHttpClientConnection;
    import org.apache.http.nio.NHttpServerConnection;
    import org.apache.http.params.HttpConnectionParams;
    import org.apache.synapse.transport.passthru.config.BaseConfiguration;
    import org.apache.synapse.transport.passthru.config.PassThroughConfiguration;
    import org.apache.synapse.transport.passthru.util.ControlledByteBuffer;

    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.nio.ByteBuffer;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;

    /**
     * This is a buffer shared by both producers and consumers.
     */
    public class Pipe {

        public static final int DEFAULT_TIME_OUT_VALUE = 180000;

        /** IOControl of the reader */
        private IOControl producerIoControl;

        /** IOControl of the consumer */
        private IOControl consumerIoControl;

        /** Fixed size buffer to read and write data */
        private ControlledByteBuffer buffer;

        private ControlledByteBuffer outputBuffer;

        private boolean producerCompleted = false;

        public boolean isProducerCompleted() {
            return producerCompleted;
        }

        /** Lock to synchronize the producers and consumers */
        private Lock lock = new ReentrantLock();

        private Condition readCondition = lock.newCondition();
        private Condition writeCondition = lock.newCondition();

        /** Name to identify the buffer */
        private String name = "Buffer";

        private boolean consumerError = false;

        private boolean producerError = false;

        /** Flag to indicate that the Pipe will not produce anything */
        private boolean discardable = false;

        /**
         * Socket Time out value specified in the nttp properties file.
         */
        private static long socketTimeOut = PassThroughConfiguration.getInstance()
                .getIntProperty(HttpConnectionParams.SO_TIMEOUT, DEFAULT_TIME_OUT_VALUE);

        /**
         * Boolean state to identify whether write condition await interrupted or timeout exceeded.
         * Normal behaviour should be the interruption of the await. If some thing went wrong and the write condition
         * did not got any notification to allow write to the buffer, the await will time out after socket time out
         * value specified in the nttp properties file.
         * "true" - if interrupted
         * "false"- if time out exceeds
         */
        private boolean awaitInterrupted = true;
        boolean isStale = false;

        private BaseConfiguration baseConfig;

        private boolean serializationComplete = false;

        private boolean rawSerializationComplete = false;

        private boolean hasHttpProducer = true;

        private ByteBufferInputStream inputStream;
        private ByteBufferOutputStream outputStream;

        public Pipe(IOControl producerIoControl, ControlledByteBuffer buffer,
                    String name, BaseConfiguration baseConfig) {
            this.producerIoControl = producerIoControl;
            this.buffer = buffer;
            this.name += "_" + name;
            this.baseConfig = baseConfig;
        }

        public Pipe(ControlledByteBuffer buffer, String name, BaseConfiguration baseConfig) {
            this.buffer = buffer;
            this.name += "_" + name;
            this.baseConfig = baseConfig;
            this.hasHttpProducer = false;
        }

        /**
         * Set the consumers IOControl
         * @param consumerIoControl IOControl of the consumer
         */
        public void attachConsumer(IOControl consumerIoControl) {
            this.consumerIoControl = consumerIoControl;
        }

        private ControlledByteBuffer getConsumerBuffer() {
            return (outputBuffer != null ? outputBuffer : buffer);
        }

        /**
         * Consume the data from the buffer. Before calling this method attachConsumer
         * method must be called with a valid IOControl.
         *
         * @param encoder encoder used to write the data means there will not be any data
         *                written in to this buffer
         * @return number of bytes written (consumed)
         * @throws IOException if an error occurred while consuming data
         */
        public int consume(final ContentEncoder encoder) throws IOException {
            if (consumerIoControl == null) {
                throw new IllegalStateException("Consumer cannot be null when calling consume");
            }

            if (hasHttpProducer && producerIoControl == null) {
                throw new IllegalStateException("Producer cannot be null when calling consume");
            }

            lock.lock();
            ControlledByteBuffer consumerBuffer = getConsumerBuffer();
            try {
                // if producer at error we have to stop the encoding and return immediately
                if (producerError) {
                    encoder.complete();
                    return -1;
                }
                setOutputMode(consumerBuffer);
                int bytesWritten = encoder.write(consumerBuffer.getByteBuffer());
                consumePostActions(consumerBuffer, encoder, bytesWritten);
                return bytesWritten;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Consume the data from the buffer. Same as {@link Pipe#consume(org.apache.http.nio.ContentEncoder)} but this gives
         * a copy of data which has been consumed.
         *
         * @param encoder encoder used to write
         * @return a buffer with data written (consumed)
         * @throws IOException if an error occurred while consuming data
         */
        public ByteBuffer copyAndConsume(final ContentEncoder encoder) throws IOException {

            if (consumerIoControl == null) {
                throw new IllegalStateException("Consumer cannot be null when calling consume");
            }
            if (hasHttpProducer && producerIoControl == null) {
                throw new IllegalStateException("Producer cannot be null when calling consume");
            }
            lock.lock();
            ControlledByteBuffer consumerBuffer = getConsumerBuffer();
            try {
                // if producer at error we have to stop the encoding and return immediately
                if (producerError) {
                    encoder.complete();
                    return null;
                }
                setOutputMode(consumerBuffer);
                // clone original buffer
                ByteBuffer originalBuffer = consumerBuffer.getByteBuffer();
                int bytesWritten = encoder.write(originalBuffer);
                ByteBuffer duplicate = originalBuffer.duplicate();
                // replicate positions of original buffer in duplicated buffer
                int position = originalBuffer.position();
                duplicate.limit(position);
                if (bytesWritten > 0) {
                    duplicate.position(position - bytesWritten);
                }
                consumePostActions(consumerBuffer, encoder, bytesWritten);
                return duplicate;
            } finally {
                lock.unlock();
            }
        }

        private void consumePostActions(ControlledByteBuffer consumerBuffer, ContentEncoder encoder, int bytesWritten)
                throws IOException {

            setInputMode(consumerBuffer);
            if (consumerBuffer.position() == 0) {
                if (outputBuffer == null) {
                    if (producerCompleted) {
                        encoder.complete();
                    } else {
                        // buffer is empty. Wait until the producer fills up the buffer
                        consumerIoControl.suspendOutput();
                    }
                } else if (serializationComplete || rawSerializationComplete) {
                    encoder.complete();
                }
            }
            if (bytesWritten > 0 && !encoder.isCompleted() && !producerCompleted && hasHttpProducer) {
                producerIoControl.requestInput();
            }
            writeCondition.signalAll();
        }

        /**
         * Produce data in to the buffer.
         *
         * @param decoder decoder to read bytes from the underlying stream
         * @return bytes read (consumed)
         * @throws IOException if an error occurs while reading data
         */
        public int produce(final ContentDecoder decoder) throws IOException {
            if (producerIoControl == null) {
                throw new IllegalStateException("Producer cannot be null when calling produce");
            }

            lock.lock();
            try {
                setInputMode(buffer);
                int bytesRead;
                try {
                    bytesRead = decoder.read(buffer.getByteBuffer());
                } catch (TruncatedChunkException ignore) {
                    // we should add the EoF character
                    buffer.putInt(-1);
                    // now the buffer's position should give us the bytes read.
                    bytesRead = buffer.position();
                }
                producePostActions(decoder);
                return bytesRead;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Same as {@link Pipe#produce(org.apache.http.nio.ContentDecoder)} and gives a copy of data produced
         *
         * @param decoder decoder to read bytes from the underlying stream
         * @return bytes of data read (consumed)
         * @throws IOException if an error occurs while reading data
         */
        public ByteBuffer copyAndProduce(final ContentDecoder decoder) throws IOException {

            if (producerIoControl == null) {
                throw new IllegalStateException("Producer cannot be null when calling produce");
            }
            lock.lock();
            try {
                ByteBuffer duplicate = null;
                setInputMode(buffer);
                int bytesRead;
                try {
                    // clone original buffer
                    ByteBuffer originalBuffer = buffer.getByteBuffer();
                    bytesRead = decoder.read(originalBuffer);
                    duplicate = originalBuffer.duplicate();
                    // replicate positions of original buffer in duplicated buffer
                    int position = originalBuffer.position();
                    duplicate.limit(position);
                    if (bytesRead > 0) {
                        duplicate.position(position - bytesRead);
                    }
                } catch (TruncatedChunkException ignore) {
                    // we should add the EoF character
                    buffer.putInt(-1);
                    duplicate.putInt(-1);
                }
                producePostActions(decoder);
                return duplicate;
            } finally {
                lock.unlock();
            }
        }

        private void producePostActions(final ContentDecoder decoder) {

            // if consumer is at error we have to let the producer complete
            if (consumerError) {
                buffer.clear();
            }
            if (!buffer.hasRemaining()) {
                // Input buffer is full. Suspend client input
                // until the origin handler frees up some space in the buffer
                producerIoControl.suspendInput();
            }
            // If there is some content in the input buffer make sure consumer output is active
            if (buffer.position() > 0 || decoder.isCompleted()) {
                if (consumerIoControl != null) {
                    consumerIoControl.requestOutput();
                }
                readCondition.signalAll();
            }
            if (decoder.isCompleted()) {
                producerCompleted = true;
            }
        }

        /**
         * Helper method to mark as producer completed. The normal behavior only set the boolean param to true only when
         * decoder is completed. However in some cases it is needed to intentionally close chunk stream and mark as
         * producer completed in order to prevent consumer further waiting on end of stream condition.
         *
         * @param decoder decoder instance to consume input
         */
        public void forceProducerComplete(final ContentDecoder decoder) {
            //no need to mark EoS if decoder is completed
            if (!decoder.isCompleted()) {
                lock.lock();
                try {
                    producerCompleted = true;
                    //there can be edge cases where consumer already  waiting on read condition ( buffer is empty )
                    //in that case marking producer complete is not enough need to clear the read condition
                    //consumer is waiting on
                    readCondition.signalAll();
                    //let consumer complete = clear consumerIoControl.suspendOutput();
                    if (consumerIoControl != null) {
                        consumerIoControl.requestOutput();
                    }
                } finally {
                    lock.unlock();
                }
            }
        }

        @Override
        public String toString() {
            return name;
        }

        public void consumerError() {
            lock.lock();
            try {
                this.consumerError = true;
                writeCondition.signalAll();
            } finally {
                lock.unlock();
            }
        }

        public void producerError() {
            lock.lock();
            try {
                this.producerError = true;
                readCondition.signalAll();
            } finally {
                lock.unlock();
            }
        }

        /**
         * Creates an InputStream object on the underlying ByteBuffer. The returned
         * InputStream can be used to read bytes from the underlying buffer which
         * is being filled by the producer.
         *
         * @return An InputStream object
         */
        public synchronized InputStream getInputStream() {
            if (inputStream == null) {
                inputStream = new ByteBufferInputStream();
            }
            return inputStream;
        }

        /**
         * Creates a separate ByteBuffer for the output data and returns an OutputStream
         * on top of it.
         *
         * @return An OutputStream object
         */
        public synchronized OutputStream getOutputStream() {
            if (outputStream == null) {
                outputBuffer = baseConfig.getBufferFactory().getBuffer();
                outputStream = new ByteBufferOutputStream();
            }
            return outputStream;
        }

        /**
         * Creates a separate ByteBuffer for the output data and returns an OutputStream
         * on top of it.
         *
         * @return An OutputStream object
         */
        public synchronized OutputStream resetOutputStream() {
            outputBuffer = baseConfig.getBufferFactory().getBuffer();
            outputStream = new ByteBufferOutputStream();
            return outputStream;
        }

        public synchronized void setSerializationComplete(boolean serializationComplete) {
            if (!this.serializationComplete) {
                this.serializationComplete = serializationComplete;
                if (consumerIoControl != null) {
                    consumerIoControl.requestOutput();
                }
            }
        }

        public void setRawSerializationComplete(boolean rawSerializationComplete) {
            this.rawSerializationComplete = rawSerializationComplete;
        }

        public void forceSetSerializationRest(){
            // If the pipe is reused to send the message out (ex. Failover endpoint retry), need to reset
            consumerError = false;
            if(this.serializationComplete){
                this.serializationComplete = false;
            }
        }

        public boolean isSerializationComplete() {
            return serializationComplete;
        }

        public ControlledByteBuffer getBuffer() {
            return buffer;
        }

        public boolean hasHttpProducer() {
            return hasHttpProducer;
        }

        private void setInputMode(ControlledByteBuffer buffer) {
            if (buffer.setInputMode()) {
                if (buffer.hasRemaining()) {
                    buffer.compact();
                } else {
                    buffer.clear();
                }
            }
        }

        private void setOutputMode(ControlledByteBuffer buffer) {
            if (buffer.setOutputMode()) {
                buffer.flip();
            }
        }

        private boolean hasData(ControlledByteBuffer buffer) {
            lock.lock();
            try {
                setOutputMode(buffer);
                return buffer.hasRemaining();
            } finally {
                lock.unlock();
            }
        }

        public boolean isDiscardable() {
            return discardable ;
        }

        public void setDiscardable(boolean discardable) {
            this.discardable  = discardable ;
        }

        /**
         * This method returns whether buffer consumption is required or not.
         *
         * @return boolean
         * @throws IOException
         */
        public boolean isConsumeRequired() throws IOException {
            lock.lock();
            boolean isInputMode = buffer.isInputMode();
            try {
                if (isInputMode) {
                    setOutputMode(buffer);
                }
                int readRemaining = buffer.remaining();
                int readPosition = buffer.position();
                setInputMode(buffer);
                int writePosition = buffer.position();
                int writeRemaining = buffer.remaining();
                // in this method we will return true when the buffer is full when reading didn't happened : writePosition == buffer.capacity() && readPosition == 0
                // we will return true if we have consumed the message partially and when there is remaining to read
                return (readRemaining == 0 && writeRemaining == readPosition) || (writePosition == buffer.capacity() && readPosition == 0);
            } finally {
                if (isInputMode) {
                    setInputMode(buffer);
                }
                lock.unlock();
            }
        }

        /**
         * Checks the state of the producerError property
         * @return boolean
         */
        public boolean isProducerError() {
            return producerError;
        }

        private class ByteBufferInputStream extends InputStream {

            @Override
            public int read() throws IOException {
                lock.lock();
                try {
                    if (!hasData(buffer)) {
                        waitForData();
                        if(producerError){
                            return -1;
                        }
                    }
                    if (isEndOfStream()) {
                        return -1;
                    }
                    return buffer.get() & 0xff;
                } finally {
                    lock.unlock();
                }
            }

            public int read(byte[] b, int off, int len) throws IOException {
                if (b == null) {
                    return 0;
                }

                lock.lock();
                try {
                    if (!hasData(buffer)) {
                        waitForData();
                        if(producerError){
                            return -1;
                        }
                    }
                    if (isEndOfStream()) {
                        return -1;
                    }
                    setOutputMode(buffer);
                    int chunk = len;
                    if (chunk > buffer.remaining()) {
                        chunk = buffer.remaining();
                    }
                    buffer.get(b, off, chunk);
                    return chunk;
                } finally {
                    lock.unlock();
                }
            }

            private void waitForData() throws IOException {
                lock.lock();
                try {
                    try {
                        while (!hasData(buffer) && !producerCompleted) {
                            if(producerError){
                                break;
                            }
                            producerIoControl.requestInput();
                            readCondition.await();
                        }
                    } catch (InterruptedException e) {
                        throw new IOException("Interrupted while waiting for data");
                    }
                } finally {
                    lock.unlock();
                }
            }

            private boolean isEndOfStream() {
                return !hasData(buffer) && producerCompleted;
            }
        }

        private class ByteBufferOutputStream extends OutputStream {

            @Override
            public void write(int b) throws IOException {
                lock.lock();
                try {
                    setInputMode(outputBuffer);
                    if (!outputBuffer.hasRemaining()) {
                        flushContent();
                        // If consume error happens inside flushContent then need to
                        // return from here.
                        if (consumerError || isStale) {
                            buffer.clear();
                            return;
                        }

                        // check whether await time out exceeded and throw an IOException
                        if(!awaitInterrupted) {
                            buffer.clear();
                            throw new IOException("Output buffer write time out exceeded");
                        }

                        setInputMode(outputBuffer);
                    }
                    outputBuffer.put((byte) b);
                } finally {
                    lock.unlock();
                }
            }

            public void write(final byte[] b, int off, int len) throws IOException {
                if (b == null) {
                    return;
                }
                lock.lock();
                try {
                    setInputMode(outputBuffer);
                    int remaining = len;
                    // if there is a consumer error or a stale connection, there is no point of trying to write.
                    // ex: when client connection is closed while writing back the response
                    if (consumerError || isStale) {
                        buffer.clear();
                        isStale = false;
                        throw new IOException("Consumer error or stale connection has occurred.");
                    }
                    if (consumerIoControl instanceof NHttpServerConnection) {
                        if (((NHttpServerConnection) consumerIoControl).isStale()) {
                            isStale = true;
                            writeCondition.signalAll();
                            buffer.clear();
                            return;
                        } else if (isProducerError()) {
                            writeCondition.signalAll();
                            buffer.clear();
                            return;
                        }
                    } else if (consumerIoControl instanceof NHttpClientConnection) {
                        if (((NHttpClientConnection) consumerIoControl).isStale()) {
                            isStale = true;
                            writeCondition.signalAll();
                            buffer.clear();
                            return;
                        }
                    }
                    while (remaining > 0 && !consumerError && !isStale && awaitInterrupted) {
                        if (!outputBuffer.hasRemaining()) {
                            flushContent();
                            // If consume error happens inside flushContent then
                            // need to return from here.
                            if (consumerError || isStale) {
                                buffer.clear();
                                break;
                            }
                            // check whether await time out exceeded and throw an IOException
                            if (!awaitInterrupted) {
                                buffer.clear();
                                throw new IOException("Output buffer write time out exceeded");
                            }
                            setInputMode(outputBuffer);
                        }
                        int chunk = Math.min(remaining, outputBuffer.remaining());
                        outputBuffer.put(b, off, chunk);
                        remaining -= chunk;
                        off += chunk;
                    }
                } finally {
                    lock.unlock();
                }
            }

            private void flushContent() throws IOException {

                if(rawSerializationComplete){
                    return;
                }

                lock.lock();

                try {
                    try {
                        while (hasData(outputBuffer) && !consumerError && !isStale) {
                            if(consumerError){
                                break;
                            }
                            if (consumerIoControl != null && writeCondition != null) {
                                consumerIoControl.requestOutput();

                                if (consumerIoControl instanceof NHttpServerConnection) {
                                    if (((NHttpServerConnection) consumerIoControl).isStale()) {
                                        isStale = true;
                                        writeCondition.signalAll();
                                    }
                                } else if (consumerIoControl instanceof NHttpClientConnection) {
                                    if (((NHttpClientConnection) consumerIoControl).isStale()) {
                                        isStale = true;
                                        writeCondition.signalAll();
                                    }
                                }
                                awaitInterrupted = writeCondition.await(socketTimeOut, TimeUnit.MILLISECONDS);
                                if(!awaitInterrupted){
                                    break;
                                }
                            }
                        }

                    } catch (InterruptedException ex) {
                        throw new IOException("Interrupted while flushing the content buffer");
                    }
                } finally {
                    lock.unlock();
                }
            }
        }

    }