/*
 * JBoss, Home of Professional Open Source.
 * Copyright 2014 Red Hat, Inc., 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 io.undertow.server.protocol.framed;

import static org.xnio.Bits.allAreClear;
import static org.xnio.Bits.allAreSet;
import static org.xnio.Bits.anyAreSet;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.TimeUnit;

import io.undertow.UndertowLogger;
import org.xnio.Buffers;
import org.xnio.ChannelListener;
import org.xnio.ChannelListeners;
import org.xnio.IoUtils;
import org.xnio.Option;
import io.undertow.connector.PooledByteBuffer;
import org.xnio.XnioExecutor;
import org.xnio.XnioIoThread;
import org.xnio.XnioWorker;
import org.xnio.channels.StreamSinkChannel;
import org.xnio.channels.StreamSourceChannel;

import io.undertow.UndertowMessages;

Source channel, used to receive framed messages.
Author:Stuart Douglas
/** * Source channel, used to receive framed messages. * * @author Stuart Douglas */
public abstract class AbstractFramedStreamSourceChannel<C extends AbstractFramedChannel<C, R, S>, R extends AbstractFramedStreamSourceChannel<C, R, S>, S extends AbstractFramedStreamSinkChannel<C, R, S>> implements StreamSourceChannel { private final ChannelListener.SimpleSetter<? extends R> readSetter = new ChannelListener.SimpleSetter(); private final ChannelListener.SimpleSetter<? extends R> closeSetter = new ChannelListener.SimpleSetter(); private final C framedChannel; private final Deque<FrameData> pendingFrameData = new LinkedList<>(); private int state = 0; private static final int STATE_DONE = 1 << 1; private static final int STATE_READS_RESUMED = 1 << 2; private static final int STATE_CLOSED = 1 << 3; private static final int STATE_LAST_FRAME = 1 << 4; private static final int STATE_IN_LISTENER_LOOP = 1 << 5; private static final int STATE_STREAM_BROKEN = 1 << 6; private static final int STATE_RETURNED_MINUS_ONE = 1 << 7; private static final int STATE_WAITNG_MINUS_ONE = 1 << 8;
The backing data for the current frame.
/** * The backing data for the current frame. */
private volatile PooledByteBuffer data; private int currentDataOriginalSize;
The amount of data left in the frame. If this is larger than the data in the backing buffer then
/** * The amount of data left in the frame. If this is larger than the data in the backing buffer then */
private long frameDataRemaining; private final Object lock = new Object(); private int waiters; private volatile boolean waitingForFrame; private int readFrameCount = 0; private long maxStreamSize = -1; private long currentStreamSize; private ChannelListener[] closeListeners = null; public AbstractFramedStreamSourceChannel(C framedChannel) { this.framedChannel = framedChannel; this.waitingForFrame = true; } public AbstractFramedStreamSourceChannel(C framedChannel, PooledByteBuffer data, long frameDataRemaining) { this.framedChannel = framedChannel; this.waitingForFrame = data == null && frameDataRemaining <= 0; this.frameDataRemaining = frameDataRemaining; this.currentStreamSize = frameDataRemaining; if (data != null) { if (!data.getBuffer().hasRemaining()) { data.close(); this.data = null; this.waitingForFrame = frameDataRemaining <= 0; } else { dataReady(null, data); } } } @Override public long transferTo(long position, long count, FileChannel target) throws IOException { if (anyAreSet(state, STATE_DONE)) { return -1; } beforeRead(); if (waitingForFrame) { return 0; } try { if (frameDataRemaining == 0 && anyAreSet(state, STATE_LAST_FRAME)) { synchronized (lock) { state |= STATE_RETURNED_MINUS_ONE; return -1; } } else if (data != null) { int old = data.getBuffer().limit(); try { if (count < data.getBuffer().remaining()) { data.getBuffer().limit((int) (data.getBuffer().position() + count)); } return target.write(data.getBuffer(), position); } finally { data.getBuffer().limit(old); decrementFrameDataRemaining(); } } return 0; } finally { exitRead(); } } private void decrementFrameDataRemaining() { if(!data.getBuffer().hasRemaining()) { frameDataRemaining -= currentDataOriginalSize; } } @Override public long transferTo(long count, ByteBuffer throughBuffer, StreamSinkChannel streamSinkChannel) throws IOException { if (anyAreSet(state, STATE_DONE)) { return -1; } beforeRead(); if (waitingForFrame) { throughBuffer.position(throughBuffer.limit()); return 0; } try { if (frameDataRemaining == 0 && anyAreSet(state, STATE_LAST_FRAME)) { synchronized (lock) { state |= STATE_RETURNED_MINUS_ONE; return -1; } } else if (data != null && data.getBuffer().hasRemaining()) { int old = data.getBuffer().limit(); try { if (count < data.getBuffer().remaining()) { data.getBuffer().limit((int) (data.getBuffer().position() + count)); } int written = streamSinkChannel.write(data.getBuffer()); if(data.getBuffer().hasRemaining()) { //we can still add more data //stick it it throughbuffer, otherwise transfer code will continue to attempt to use this method throughBuffer.clear(); Buffers.copy(throughBuffer, data.getBuffer()); throughBuffer.flip(); } else { throughBuffer.position(throughBuffer.limit()); } return written; } finally { data.getBuffer().limit(old); decrementFrameDataRemaining(); } } else { throughBuffer.position(throughBuffer.limit()); } return 0; } finally { exitRead(); } } public long getMaxStreamSize() { return maxStreamSize; } public void setMaxStreamSize(long maxStreamSize) { this.maxStreamSize = maxStreamSize; if(maxStreamSize > 0) { if(maxStreamSize < currentStreamSize) { handleStreamTooLarge(); } } } private void handleStreamTooLarge() { IoUtils.safeClose(this); } @Override public void suspendReads() { synchronized (lock) { state &= ~STATE_READS_RESUMED; } }
Method that is invoked when all data has been read.
Throws:
  • IOException –
/** * Method that is invoked when all data has been read. * * @throws IOException */
protected void complete() throws IOException { close(); } protected boolean isComplete() { return anyAreSet(state, STATE_DONE); } @Override public void resumeReads() { resumeReadsInternal(false); } @Override public boolean isReadResumed() { return anyAreSet(state, STATE_READS_RESUMED); } @Override public void wakeupReads() { resumeReadsInternal(true); } public void addCloseTask(ChannelListener<R> channelListener) { if(closeListeners == null) { closeListeners = new ChannelListener[]{channelListener}; } else { ChannelListener[] old = closeListeners; closeListeners = new ChannelListener[old.length + 1]; System.arraycopy(old, 0, closeListeners, 0, old.length); closeListeners[old.length] = channelListener; } }
For this class there is no difference between a resume and a wakeup
/** * For this class there is no difference between a resume and a wakeup */
void resumeReadsInternal(boolean wakeup) { synchronized (lock) { boolean alreadyResumed = anyAreSet(state, STATE_READS_RESUMED); state |= STATE_READS_RESUMED; if (!alreadyResumed || wakeup) { if (!anyAreSet(state, STATE_IN_LISTENER_LOOP)) { state |= STATE_IN_LISTENER_LOOP; getFramedChannel().runInIoThread(new Runnable() { @Override public void run() { try { boolean moreData; do { ChannelListener<? super R> listener = getReadListener(); if (listener == null || !isReadResumed()) { return; } ChannelListeners.invokeChannelListener((R) AbstractFramedStreamSourceChannel.this, listener); //if writes are shutdown or we become active then we stop looping //we stop when writes are shutdown because we can't flush until we are active //although we may be flushed as part of a batch moreData = (frameDataRemaining > 0 && data != null) || !pendingFrameData.isEmpty() || anyAreSet(state, STATE_WAITNG_MINUS_ONE); } while (allAreSet(state, STATE_READS_RESUMED) && allAreClear(state, STATE_CLOSED | STATE_STREAM_BROKEN) && moreData); } finally { state &= ~STATE_IN_LISTENER_LOOP; } } }); } } } } private ChannelListener<? super R> getReadListener() { return (ChannelListener<? super R>) readSetter.get(); } @Override public void shutdownReads() throws IOException { close(); } protected void lastFrame() { synchronized (lock) { state |= STATE_LAST_FRAME; } waitingForFrame = false; if(data == null && pendingFrameData.isEmpty() && frameDataRemaining == 0) { state |= STATE_DONE | STATE_CLOSED; getFramedChannel().notifyFrameReadComplete(this); IoUtils.safeClose(this); } } protected boolean isLastFrame() { return anyAreSet(state, STATE_LAST_FRAME); } @Override public void awaitReadable() throws IOException { if(Thread.currentThread() == getIoThread()) { throw UndertowMessages.MESSAGES.awaitCalledFromIoThread(); } if (data == null && pendingFrameData.isEmpty()) { synchronized (lock) { if (data == null && pendingFrameData.isEmpty()) { try { waiters++; lock.wait(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new InterruptedIOException(); } finally { waiters--; } } } } } @Override public void awaitReadable(long l, TimeUnit timeUnit) throws IOException { if(Thread.currentThread() == getIoThread()) { throw UndertowMessages.MESSAGES.awaitCalledFromIoThread(); } if (data == null) { synchronized (lock) { if (data == null) { try { waiters++; lock.wait(timeUnit.toMillis(l)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new InterruptedIOException(); } finally { waiters--; } } } } }
Called when data has been read from the underlying channel.
Params:
  • headerData – The frame header data. This may be null if the data is part of a an existing frame
  • frameData – The frame data
/** * Called when data has been read from the underlying channel. * * @param headerData The frame header data. This may be null if the data is part of a an existing frame * @param frameData The frame data */
protected void dataReady(FrameHeaderData headerData, PooledByteBuffer frameData) { if(anyAreSet(state, STATE_STREAM_BROKEN | STATE_CLOSED)) { frameData.close(); return; } synchronized (lock) { boolean newData = pendingFrameData.isEmpty(); this.pendingFrameData.add(new FrameData(headerData, frameData)); if (newData) { if (waiters > 0) { lock.notifyAll(); } } waitingForFrame = false; } if (anyAreSet(state, STATE_READS_RESUMED)) { resumeReadsInternal(true); } if(headerData != null) { currentStreamSize += headerData.getFrameLength(); if(maxStreamSize > 0 && currentStreamSize > maxStreamSize) { handleStreamTooLarge(); } } } protected long updateFrameDataRemaining(PooledByteBuffer frameData, long frameDataRemaining) { return frameDataRemaining; } protected PooledByteBuffer processFrameData(PooledByteBuffer data, boolean lastFragmentOfFrame) throws IOException { return data; } protected void handleHeaderData(FrameHeaderData headerData) { } @Override public XnioExecutor getReadThread() { return framedChannel.getIoThread(); } @Override public ChannelListener.Setter<? extends R> getReadSetter() { return readSetter; } @Override public ChannelListener.Setter<? extends R> getCloseSetter() { return closeSetter; } @Override public XnioWorker getWorker() { return framedChannel.getWorker(); } @Override public XnioIoThread getIoThread() { return framedChannel.getIoThread(); } @Override public boolean supportsOption(Option<?> option) { return false; } @Override public <T> T getOption(Option<T> tOption) throws IOException { return null; } @Override public <T> T setOption(Option<T> tOption, T t) throws IllegalArgumentException, IOException { return null; } @Override public long read(ByteBuffer[] dsts, int offset, int length) throws IOException { if (anyAreSet(state, STATE_DONE)) { return -1; } beforeRead(); if (waitingForFrame) { return 0; } try { if (frameDataRemaining == 0 && anyAreSet(state, STATE_LAST_FRAME)) { synchronized (lock) { state |= STATE_RETURNED_MINUS_ONE; } return -1; } else if (data != null) { int old = data.getBuffer().limit(); try { long count = Buffers.remaining(dsts, offset, length); if (count < data.getBuffer().remaining()) { data.getBuffer().limit((int) (data.getBuffer().position() + count)); } else { count = data.getBuffer().remaining(); } return Buffers.copy((int) count, dsts, offset, length, data.getBuffer()); } finally { data.getBuffer().limit(old); decrementFrameDataRemaining(); } } return 0; } finally { exitRead(); } } @Override public long read(ByteBuffer[] dsts) throws IOException { return read(dsts, 0, dsts.length); } @Override public int read(ByteBuffer dst) throws IOException { if (anyAreSet(state, STATE_DONE)) { return -1; } if (!dst.hasRemaining()) { return 0; } beforeRead(); if (waitingForFrame) { return 0; } try { if (frameDataRemaining == 0 && anyAreSet(state, STATE_LAST_FRAME)) { synchronized (lock) { state |= STATE_RETURNED_MINUS_ONE; } return -1; } else if (data != null) { int old = data.getBuffer().limit(); try { int count = dst.remaining(); if (count < data.getBuffer().remaining()) { data.getBuffer().limit(data.getBuffer().position() + count); } else { count = data.getBuffer().remaining(); } return Buffers.copy(count, dst, data.getBuffer()); } finally { data.getBuffer().limit(old); decrementFrameDataRemaining(); } } return 0; } finally { try { exitRead(); } catch (Throwable e) { markStreamBroken(); } } } private void beforeRead() throws IOException { if (anyAreSet(state, STATE_STREAM_BROKEN)) { throw UndertowMessages.MESSAGES.channelIsClosed(); } if (data == null) { synchronized (lock) { FrameData pending = pendingFrameData.poll(); if (pending != null) { PooledByteBuffer frameData = pending.getFrameData(); boolean hasData = true; if(!frameData.getBuffer().hasRemaining()) { frameData.close(); hasData = false; } if (pending.getFrameHeaderData() != null) { this.frameDataRemaining = pending.getFrameHeaderData().getFrameLength(); handleHeaderData(pending.getFrameHeaderData()); } if(hasData) { this.frameDataRemaining = updateFrameDataRemaining(frameData, frameDataRemaining); this.currentDataOriginalSize = frameData.getBuffer().remaining(); try { this.data = processFrameData(frameData, frameDataRemaining - currentDataOriginalSize == 0); } catch (Throwable e) { frameData.close(); UndertowLogger.REQUEST_IO_LOGGER.ioException(new IOException(e)); markStreamBroken(); } } } } } } private void exitRead() throws IOException { if (data != null && !data.getBuffer().hasRemaining()) { data.close(); data = null; } if (frameDataRemaining == 0) { try { synchronized (lock) { readFrameCount++; if (pendingFrameData.isEmpty()) { if (anyAreSet(state, STATE_RETURNED_MINUS_ONE)) { state |= STATE_DONE; complete(); close(); } else if(anyAreSet(state, STATE_LAST_FRAME)) { state |= STATE_WAITNG_MINUS_ONE; } else { waitingForFrame = true; } } } } finally { if (pendingFrameData.isEmpty()) { framedChannel.notifyFrameReadComplete(this); } } } } @Override public boolean isOpen() { return allAreClear(state, STATE_CLOSED); } @Override public void close() { if(anyAreSet(state, STATE_CLOSED)) { return; } synchronized (lock) { state |= STATE_CLOSED; if (allAreClear(state, STATE_DONE | STATE_LAST_FRAME)) { state |= STATE_STREAM_BROKEN; channelForciblyClosed(); } if (data != null) { data.close(); data = null; } while (!pendingFrameData.isEmpty()) { pendingFrameData.poll().frameData.close(); } ChannelListeners.invokeChannelListener(this, (ChannelListener<? super AbstractFramedStreamSourceChannel<C, R, S>>) closeSetter.get()); if (closeListeners != null) { for (int i = 0; i < closeListeners.length; ++i) { closeListeners[i].handleEvent(this); } } } } protected void channelForciblyClosed() { //TODO: what should be the default action? //we can probably just ignore it, as it does not affect the underlying protocol } protected C getFramedChannel() { return framedChannel; } protected int getReadFrameCount() { return readFrameCount; }
Called when this stream is no longer valid. Reads from the stream will result in an exception.
/** * Called when this stream is no longer valid. Reads from the stream will result * in an exception. */
protected void markStreamBroken() { if(anyAreSet(state, STATE_STREAM_BROKEN)) { return; } synchronized (lock) { state |= STATE_STREAM_BROKEN; PooledByteBuffer data = this.data; if(data != null) { try { data.close(); //may have been closed by the read thread } catch (Throwable e) { //ignore } this.data = null; } for(FrameData frame : pendingFrameData) { frame.frameData.close(); } pendingFrameData.clear(); if(isReadResumed()) { resumeReadsInternal(true); } if (waiters > 0) { lock.notifyAll(); } } } private class FrameData { private final FrameHeaderData frameHeaderData; private final PooledByteBuffer frameData; FrameData(FrameHeaderData frameHeaderData, PooledByteBuffer frameData) { this.frameHeaderData = frameHeaderData; this.frameData = frameData; } FrameHeaderData getFrameHeaderData() { return frameHeaderData; } PooledByteBuffer getFrameData() { return frameData; } } }