/*
 * JBoss, Home of Professional Open Source.
 *
 * Copyright 2012 Red Hat, Inc. and/or its affiliates, and individual
 * contributors as indicated by the @author tags.
 *
 * 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.xnio.channels;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.TimeUnit;
import org.xnio.Buffers;
import org.xnio.ChannelListener;
import org.xnio.ChannelListeners;
import org.xnio.Option;
import org.xnio.Pooled;
import org.xnio.XnioExecutor;
import org.xnio.XnioIoThread;
import org.xnio.XnioWorker;

A stream source channel which can have data pushed back into it.  Note that waiting readers will NOT be interrupted
when data is pushed back; therefore data should only be pushed back at points when no waiters are expected to exist.
Author:
David M. Lloyd/**
 * A stream source channel which can have data pushed back into it.  Note that waiting readers will NOT be interrupted
 * when data is pushed back; therefore data should only be pushed back at points when no waiters are expected to exist.
 *
 * @author <a href="mailto:david.lloyd@redhat.com">David M. Lloyd</a>
 */
public final class PushBackStreamChannel implements StreamSourceChannel, WrappedChannel<StreamSourceChannel> {

    private final StreamSourceChannel firstChannel;
    private StreamSourceChannel channel;

    private ChannelListener<? super PushBackStreamChannel> readListener;
    private ChannelListener<? super PushBackStreamChannel> closeListener;

    
Construct a new instance.
Params:
channel – the channel to wrap/**
     * Construct a new instance.
     *
     * @param channel the channel to wrap
     */
    public PushBackStreamChannel(final StreamSourceChannel channel) {
        this.channel = firstChannel = channel;
        firstChannel.getReadSetter().set(new ChannelListener<StreamSourceChannel>() {
            public void handleEvent(final StreamSourceChannel channel) {
                ChannelListeners.invokeChannelListener(PushBackStreamChannel.this, readListener);
            }
        });
        firstChannel.getCloseSetter().set(new ChannelListener<StreamSourceChannel>() {
            public void handleEvent(final StreamSourceChannel channel) {
                ChannelListeners.invokeChannelListener(PushBackStreamChannel.this, closeListener);
            }
        });
    }

    public void setReadListener(final ChannelListener<? super PushBackStreamChannel> readListener) {
        this.readListener = readListener;
    }

    public void setCloseListener(final ChannelListener<? super PushBackStreamChannel> closeListener) {
        this.closeListener = closeListener;
    }

    public ChannelListener.Setter<? extends PushBackStreamChannel> getReadSetter() {
        return new ChannelListener.Setter<PushBackStreamChannel>() {
            public void set(final ChannelListener<? super PushBackStreamChannel> listener) {
                setReadListener(listener);
            }
        };
    }

    public ChannelListener.Setter<? extends PushBackStreamChannel> getCloseSetter() {
        return new ChannelListener.Setter<PushBackStreamChannel>() {
            public void set(final ChannelListener<? super PushBackStreamChannel> listener) {
                setCloseListener(listener);
            }
        };
    }

    public long transferTo(final long position, final long count, final FileChannel target) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel == null) {
            return 0;
        }
        return channel.transferTo(position, count, target);
    }

    public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel == null) {
            return -1L;
        }
        return channel.transferTo(count, throughBuffer, target);
    }

    public int read(final ByteBuffer dst) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel == null) {
            return -1;
        }
        return channel.read(dst);
    }

    public long read(final ByteBuffer[] dsts) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel == null) {
            return -1L;
        }
        return channel.read(dsts);
    }

    public long read(final ByteBuffer[] dsts, final int offset, final int length) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel == null) {
            return -1L;
        }
        return channel.read(dsts, offset, length);
    }

    
Re-queue the given pooled buffer into this channel.  This method transfers ownership of the given buffer
to this channel.  The buffer should be flipped for emptying.
Params:
buffer – the buffer to re-queue/**
     * Re-queue the given pooled buffer into this channel.  This method transfers ownership of the given buffer
     * to this channel.  The buffer should be flipped for emptying.
     *
     * @param buffer the buffer to re-queue
     */
    public void unget(Pooled<ByteBuffer> buffer) {
        StreamSourceChannel old;
        old = channel;
        if (old == null) {
            buffer.free();
            return;
        }
        channel = new BufferHolder(old, buffer);
    }

    public void suspendReads() {
        firstChannel.suspendReads();
    }

    public void resumeReads() {
        final StreamSourceChannel channel = this.channel;
        if (channel != null) {
            channel.resumeReads();
        }
    }

    public boolean isReadResumed() {
        return firstChannel.isReadResumed();
    }

    public void wakeupReads() {
        firstChannel.wakeupReads();
    }

    public void shutdownReads() throws IOException {
        final StreamSourceChannel old = channel;
        if (old != null) {
            channel = null;
            old.shutdownReads();
        }
    }

    public void awaitReadable() throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel != null) {
            channel.awaitReadable();
        }
    }

    public void awaitReadable(final long time, final TimeUnit timeUnit) throws IOException {
        final StreamSourceChannel channel = this.channel;
        if (channel != null) {
            channel.awaitReadable(time, timeUnit);
        }
    }

    @Deprecated
    public XnioExecutor getReadThread() {
        return firstChannel.getReadThread();
    }

    public XnioIoThread getIoThread() {
        return firstChannel.getIoThread();
    }

    public XnioWorker getWorker() {
        return firstChannel.getWorker();
    }

    public boolean isOpen() {
        return firstChannel.isOpen();
    }

    public void close() throws IOException {
        final StreamSourceChannel old = channel;
        if (old != null) {
            channel = null;
            old.close();
        }
    }

    public boolean supportsOption(final Option<?> option) {
        return firstChannel.supportsOption(option);
    }

    public <T> T getOption(final Option<T> option) throws IOException {
        return firstChannel.getOption(option);
    }

    public <T> T setOption(final Option<T> option, final T value) throws IllegalArgumentException, IOException {
        return firstChannel.setOption(option, value);
    }

    public StreamSourceChannel getChannel() {
        return firstChannel;
    }

    class BufferHolder implements StreamSourceChannel {
        private final StreamSourceChannel next;
        private final Pooled<ByteBuffer> buffer;

        BufferHolder(final StreamSourceChannel next, final Pooled<ByteBuffer> buffer) {
            this.next = next;
            this.buffer = buffer;
        }

        public long transferTo(long position, long count, FileChannel target) throws IOException {
            long cnt;
            final ByteBuffer src;
            try {
                src = buffer.getResource();
                final int pos = src.position();
                final int rem = src.remaining();
                if (rem > count) try {
                    // partial empty of our buffer
                    src.limit(pos + (int) count);
                    return target.write(src, position);
                } finally {
                    src.limit(pos + rem);
                } else {
                    // full empty of our buffer
                    cnt = target.write(src, position);
                    if (cnt == rem) {
                        // we emptied our buffer
                        channel = next;
                        buffer.free();
                    } else {
                        return cnt;
                    }
                    position += cnt;
                    count -= cnt;
                }
            } catch (IllegalStateException ignored) {
                channel = next;
                cnt = 0L;
            }
            return cnt + next.transferTo(position, count, target);
        }

        public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException {
            long cnt;
            throughBuffer.clear();
            final ByteBuffer src;
            try {
                src = buffer.getResource();
                final int pos = src.position();
                final int rem = src.remaining();
                if (rem > count) try {
                    // partial empty of our buffer
                    src.limit(pos + (int) count);
                    throughBuffer.limit(0);
                    return target.write(src);
                } finally {
                    src.limit(pos + rem);
                } else {
                    // full empty of our buffer
                    cnt = target.write(src);
                    if (cnt == rem) {
                        // we emptied our buffer
                        channel = next;
                        buffer.free();
                    } else {
                        return cnt;
                    }
                }
            } catch (IllegalStateException ignored) {
                channel = next;
                cnt = 0L;
            }
            final long res = next.transferTo(count - cnt, throughBuffer, target);
            return res > 0L ? cnt + res : cnt > 0L ? cnt : res;
        }

        public long read(final ByteBuffer[] dsts, final int offset, final int length) throws IOException {
            long cnt;
            try {
                final ByteBuffer src = buffer.getResource();
                cnt = Buffers.copy(dsts, offset, length, src);
                if (src.hasRemaining()) {
                    return cnt;
                }
                final StreamSourceChannel next = channel = this.next;
                buffer.free();
                if (cnt > 0L && next == firstChannel) {
                    // don't hit the main channel until the user wants to
                    return cnt;
                }
            } catch (IllegalStateException ignored) {
                channel = next;
                cnt = 0;
            }
            final long res = next.read(dsts, offset, length);
            return res > 0 ? res + cnt : cnt > 0 ? cnt : res;
        }

        public long read(final ByteBuffer[] dsts) throws IOException {
            return read(dsts, 0, dsts.length);
        }

        public int read(final ByteBuffer dst) throws IOException {
            int cnt;
            if (! dst.hasRemaining()) {
                return 0;
            }
            try {
                final ByteBuffer src = buffer.getResource();
                cnt = Buffers.copy(dst, src);
                if (src.hasRemaining()) {
                    return cnt;
                }
                final StreamSourceChannel next = channel = this.next;
                buffer.free();
                if (cnt > 0 && next == firstChannel) {
                    // don't hit the main channel until the user wants to
                    return cnt;
                }
            } catch (IllegalStateException ignored) {
                channel = next;
                cnt = 0;
            }
            final int res = next.read(dst);
            return res > 0 ? res + cnt : cnt > 0 ? cnt : res;
        }

        public void close() throws IOException {
            buffer.free();
            next.close();
        }

        public void resumeReads() {
            // reads are always ready in this case
            firstChannel.wakeupReads();
        }

        public void shutdownReads() throws IOException {
            buffer.free();
            next.shutdownReads();
        }

        public void awaitReadable() throws IOException {
            // return immediately
        }

        public void awaitReadable(final long time, final TimeUnit timeUnit) throws IOException {
            // return immediately
        }

        // unused methods

        public boolean isOpen() {
            throw new UnsupportedOperationException();
        }

        public ChannelListener.Setter<? extends StreamSourceChannel> getReadSetter() {
            throw new UnsupportedOperationException();
        }

        public ChannelListener.Setter<? extends StreamSourceChannel> getCloseSetter() {
            throw new UnsupportedOperationException();
        }

        public void suspendReads() {
            throw new UnsupportedOperationException();
        }

        public boolean isReadResumed() {
            throw new UnsupportedOperationException();
        }

        public void wakeupReads() {
            throw new UnsupportedOperationException();
        }

        @Deprecated
        public XnioExecutor getReadThread() {
            throw new UnsupportedOperationException();
        }

        public XnioIoThread getIoThread() {
            throw new UnsupportedOperationException();
        }

        public XnioWorker getWorker() {
            throw new UnsupportedOperationException();
        }

        public boolean supportsOption(final Option<?> option) {
            throw new UnsupportedOperationException();
        }

        public <T> T getOption(final Option<T> option) throws IOException {
            throw new UnsupportedOperationException();
        }

        public <T> T setOption(final Option<T> option, final T value) throws IllegalArgumentException, IOException {
            throw new UnsupportedOperationException();
        }
    }
}