/*
 * 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.conduits;

import io.undertow.UndertowLogger;
import org.xnio.Buffers;
import org.xnio.channels.FixedLengthOverflowException;
import org.xnio.channels.StreamSourceChannel;
import org.xnio.conduits.AbstractStreamSinkConduit;
import org.xnio.conduits.Conduits;
import org.xnio.conduits.StreamSinkConduit;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.concurrent.TimeUnit;

import static java.lang.Math.min;
import static org.xnio.Bits.allAreClear;
import static org.xnio.Bits.allAreSet;
import static org.xnio.Bits.anyAreSet;
import static org.xnio.Bits.longBitMask;

A channel which writes a fixed amount of data. A listener is called once the data has been written.
Author:David M. Lloyd
/** * A channel which writes a fixed amount of data. A listener is called once the data has been written. * * @author <a href="mailto:david.lloyd@redhat.com">David M. Lloyd</a> */
public abstract class AbstractFixedLengthStreamSinkConduit extends AbstractStreamSinkConduit<StreamSinkConduit> { private int config; private long state; private boolean broken = false; private static final int CONF_FLAG_CONFIGURABLE = 1 << 0; private static final int CONF_FLAG_PASS_CLOSE = 1 << 1; private static final long FLAG_CLOSE_REQUESTED = 1L << 63L; private static final long FLAG_CLOSE_COMPLETE = 1L << 62L; private static final long FLAG_FINISHED_CALLED = 1L << 61L; private static final long MASK_COUNT = longBitMask(0, 60);
Construct a new instance.
Params:
  • next – the next channel
  • contentLength – the content length
  • configurable – true if this instance should pass configuration to the next
  • propagateClose – true if this instance should pass close to the next
/** * Construct a new instance. * * @param next the next channel * @param contentLength the content length * @param configurable {@code true} if this instance should pass configuration to the next * @param propagateClose {@code true} if this instance should pass close to the next */
public AbstractFixedLengthStreamSinkConduit(final StreamSinkConduit next, final long contentLength, final boolean configurable, final boolean propagateClose) { super(next); if (contentLength < 0L) { throw new IllegalArgumentException("Content length must be greater than or equal to zero"); } else if (contentLength > MASK_COUNT) { throw new IllegalArgumentException("Content length is too long"); } config = (configurable ? CONF_FLAG_CONFIGURABLE : 0) | (propagateClose ? CONF_FLAG_PASS_CLOSE : 0); this.state = contentLength; } protected void reset(long contentLength, boolean propagateClose) { this.state = contentLength; if (propagateClose) { config |= CONF_FLAG_PASS_CLOSE; } else { config &= ~CONF_FLAG_PASS_CLOSE; } } public int write(final ByteBuffer src) throws IOException { long val = state; final long remaining = val & MASK_COUNT; if (!src.hasRemaining()) { return 0; } if (allAreSet(val, FLAG_CLOSE_REQUESTED)) { throw new ClosedChannelException(); } int oldLimit = src.limit(); if (remaining == 0) { throw new FixedLengthOverflowException(); } else if (src.remaining() > remaining) { src.limit((int) (src.position() + remaining)); } int res = 0; try { return res = next.write(src); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } finally { src.limit(oldLimit); exitWrite(val, (long) res); } } public long write(final ByteBuffer[] srcs, final int offset, final int length) throws IOException { if (length == 0) { return 0L; } else if (length == 1) { return write(srcs[offset]); } long val = state; final long remaining = val & MASK_COUNT; if (allAreSet(val, FLAG_CLOSE_REQUESTED)) { throw new ClosedChannelException(); } long toWrite = Buffers.remaining(srcs, offset, length); if (remaining == 0) { throw new FixedLengthOverflowException(); } int[] limits = null; if (toWrite > remaining) { limits = new int[length]; long r = remaining; for (int i = offset; i < offset + length; ++i) { limits[i - offset] = srcs[i].limit(); int br = srcs[i].remaining(); if(br < r) { r -= br; } else { srcs[i].limit((int) (srcs[i].position() + r)); r = 0; } } } long res = 0L; try { return res = next.write(srcs, offset, length); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } finally { if (limits != null) { for (int i = offset; i < offset + length; ++i) { srcs[i].limit(limits[i - offset]); } } exitWrite(val, res); } } @Override public long writeFinal(ByteBuffer[] srcs, int offset, int length) throws IOException { try { return Conduits.writeFinalBasic(this, srcs, offset, length); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } } @Override public int writeFinal(ByteBuffer src) throws IOException { try { return Conduits.writeFinalBasic(this, src); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } } public long transferFrom(final FileChannel src, final long position, final long count) throws IOException { if (count == 0L) return 0L; long val = state; if (allAreSet(val, FLAG_CLOSE_REQUESTED)) { throw new ClosedChannelException(); } if (allAreClear(val, MASK_COUNT)) { throw new FixedLengthOverflowException(); } long res = 0L; try { return res = next.transferFrom(src, position, min(count, (val & MASK_COUNT))); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } finally { exitWrite(val, res); } } public long transferFrom(final StreamSourceChannel source, final long count, final ByteBuffer throughBuffer) throws IOException { if (count == 0L) return 0L; long val = state; if (allAreSet(val, FLAG_CLOSE_REQUESTED)) { throw new ClosedChannelException(); } if (allAreClear(val, MASK_COUNT)) { throw new FixedLengthOverflowException(); } long res = 0L; try { return res = next.transferFrom(source, min(count, (val & MASK_COUNT)), throughBuffer); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } finally { exitWrite(val, res); } } public boolean flush() throws IOException { long val = state; if (anyAreSet(val, FLAG_CLOSE_COMPLETE)) { return true; } boolean flushed = false; try { return flushed = next.flush(); } catch (IOException | RuntimeException | Error e) { broken = true; throw e; } finally { exitFlush(val, flushed); } } public boolean isWriteResumed() { // not perfect but not provably wrong either... return allAreClear(state, FLAG_CLOSE_COMPLETE) && next.isWriteResumed(); } public void wakeupWrites() { long val = state; if (anyAreSet(val, FLAG_CLOSE_COMPLETE)) { return; } next.wakeupWrites(); } public void terminateWrites() throws IOException { final long val = enterShutdown(); if (anyAreSet(val, MASK_COUNT) && !broken) { UndertowLogger.REQUEST_IO_LOGGER.debugf("Fixed length stream closed with with %s bytes remaining", val & MASK_COUNT); try { next.truncateWrites(); } finally { if (!anyAreSet(state, FLAG_FINISHED_CALLED)) { state |= FLAG_FINISHED_CALLED; channelFinished(); } } } else if (allAreSet(config, CONF_FLAG_PASS_CLOSE)) { next.terminateWrites(); } } @Override public void truncateWrites() throws IOException { try { if (!anyAreSet(state, FLAG_FINISHED_CALLED)) { state |= FLAG_FINISHED_CALLED; channelFinished(); } } finally { super.truncateWrites(); } } public void awaitWritable() throws IOException { next.awaitWritable(); } public void awaitWritable(final long time, final TimeUnit timeUnit) throws IOException { next.awaitWritable(time, timeUnit); }
Get the number of remaining bytes in this fixed length channel.
Returns:the number of remaining bytes
/** * Get the number of remaining bytes in this fixed length channel. * * @return the number of remaining bytes */
public long getRemaining() { return state & MASK_COUNT; } private void exitWrite(long oldVal, long consumed) { long newVal = oldVal - consumed; state = newVal; } private void exitFlush(long oldVal, boolean flushed) { long newVal = oldVal; boolean callFinish = false; if ((anyAreSet(oldVal, FLAG_CLOSE_REQUESTED) || (newVal & MASK_COUNT) == 0L) && flushed) { newVal |= FLAG_CLOSE_COMPLETE; if (!anyAreSet(oldVal, FLAG_FINISHED_CALLED) && (newVal & MASK_COUNT) == 0L) { newVal |= FLAG_FINISHED_CALLED; callFinish = true; } state = newVal; if (callFinish) { channelFinished(); } } } protected void channelFinished() { } private long enterShutdown() { long oldVal, newVal; oldVal = state; if (anyAreSet(oldVal, FLAG_CLOSE_REQUESTED | FLAG_CLOSE_COMPLETE)) { // no action necessary return oldVal; } newVal = oldVal | FLAG_CLOSE_REQUESTED; if (anyAreSet(oldVal, MASK_COUNT)) { // error: channel not filled. set both close flags. newVal |= FLAG_CLOSE_COMPLETE; } state = newVal; return oldVal; } }