/*
 * JBoss, Home of Professional Open Source
 *
 * Copyright 2013 Red Hat, Inc. and/or its affiliates.
 *
 * 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.conduits;

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.Pooled;
import org.xnio.channels.StreamSinkChannel;

A stream source conduit which allows buffers to be "pushed back" to the head of the stream.
Author:David M. Lloyd
/** * A stream source conduit which allows buffers to be "pushed back" to the head of the stream. * * @author <a href="mailto:david.lloyd@redhat.com">David M. Lloyd</a> */
public final class PushBackStreamSourceConduit extends AbstractStreamSourceConduit<StreamSourceConduit> implements StreamSourceConduit { private StreamSourceConduit current = next; private boolean shutdown;
Construct a new instance.
Params:
  • next – the delegate conduit to set
/** * Construct a new instance. * * @param next the delegate conduit to set */
public PushBackStreamSourceConduit(final StreamSourceConduit next) { super(next); } public void resumeReads() { current.resumeReads(); } public int read(final ByteBuffer dst) throws IOException { return current.read(dst); } public long read(final ByteBuffer[] dsts, final int offs, final int len) throws IOException { return current.read(dsts, offs, len); } public long transferTo(final long position, final long count, final FileChannel target) throws IOException { return current.transferTo(position, count, target); } public long transferTo(final long count, final ByteBuffer throughBuffer, final StreamSinkChannel target) throws IOException { return current.transferTo(count, throughBuffer, target); } public void awaitReadable() throws IOException { current.awaitReadable(); } public void awaitReadable(final long time, final TimeUnit timeUnit) throws IOException { current.awaitReadable(time, timeUnit); } public void terminateReads() throws IOException { shutdown = true; current.terminateReads(); } public void setReadReadyHandler(final ReadReadyHandler handler) { current.setReadReadyHandler(handler); }
Push a buffer back to the head of the stream. Once the buffer data is consumed, it will be released back to its original pool (if any).
Params:
  • pooledBuffer – the buffer to push back
/** * Push a buffer back to the head of the stream. Once the buffer data is consumed, it will * be released back to its original pool (if any). * * @param pooledBuffer the buffer to push back */
public void pushBack(Pooled<ByteBuffer> pooledBuffer) { if (pooledBuffer == null) { return; } if (shutdown || ! pooledBuffer.getResource().hasRemaining()) { pooledBuffer.free(); } else { current = new BufferConduit(current, pooledBuffer); } } class BufferConduit extends AbstractStreamSourceConduit<StreamSourceConduit> implements StreamSourceConduit { private final Pooled<ByteBuffer> pooledBuffer; BufferConduit(final StreamSourceConduit next, final Pooled<ByteBuffer> pooledBuffer) { super(next); this.pooledBuffer = pooledBuffer; } public void resumeReads() { next.wakeupReads(); } public void awaitReadable(final long time, final TimeUnit timeUnit) throws IOException { // readable } public void awaitReadable() throws IOException { // readable } public int read(final ByteBuffer dst) throws IOException { int cnt; if (! dst.hasRemaining()) { return 0; } final StreamSourceConduit next = this.next; try { final ByteBuffer src = pooledBuffer.getResource(); cnt = Buffers.copy(dst, src); if (src.hasRemaining()) { return cnt; } current = next; pooledBuffer.free(); if (cnt > 0 && next == PushBackStreamSourceConduit.this.next) { // don't hit the main channel until the user wants to return cnt; } } catch (IllegalStateException ignored) { current = next; cnt = 0; } final int res = next.read(dst); return res > 0 ? res + cnt : cnt > 0 ? cnt : res; } public long read(final ByteBuffer[] dsts, final int offs, final int len) throws IOException { long cnt; final StreamSourceConduit next = this.next; try { final ByteBuffer src = pooledBuffer.getResource(); cnt = Buffers.copy(dsts, offs, len, src); if (src.hasRemaining()) { return cnt; } current = next; pooledBuffer.free(); if (cnt > 0L && next == PushBackStreamSourceConduit.this.next) { // don't hit the main channel until the user wants to return cnt; } } catch (IllegalStateException ignored) { current = next; cnt = 0; } final long res = next.read(dsts, offs, len); return res > 0 ? res + cnt : cnt > 0 ? cnt : res; } public long transferTo(long position, long count, final FileChannel target) throws IOException { long cnt; final ByteBuffer src; try { src = pooledBuffer.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 current = next; pooledBuffer.free(); } else { return cnt; } position += cnt; count -= cnt; } } catch (IllegalStateException ignored) { current = 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; final ByteBuffer src; try { src = pooledBuffer.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); int res = target.write(src); if(res == 0) { //a bit yuck, but if we have filed to copy anything we need to transfer data into the throughbuffer //this signals to the called that it was not the read that did not succeed, but the write throughBuffer.clear(); Buffers.copy(throughBuffer, src); throughBuffer.flip(); } else { //make sure throughbuffer is empty throughBuffer.clear(); throughBuffer.flip(); } return res; } finally { src.limit(pos + rem); } else { // full empty of our buffer cnt = target.write(src); if (cnt == rem) { // we emptied our buffer current = next; pooledBuffer.free(); } else { if (cnt == 0) { //a bit yuck, but if we have filed to copy anything we need to transfer data into the throughbuffer //this signals to the called that it was not the read that did not succeed, but the write throughBuffer.clear(); Buffers.copy(throughBuffer, src); throughBuffer.flip(); } else { //make sure throughbuffer is empty throughBuffer.clear(); throughBuffer.flip(); } return cnt; } } } catch (IllegalStateException ignored) { current = next; cnt = 0L; } final long res = next.transferTo(count - cnt, throughBuffer, target); return res > 0L ? cnt + res : cnt > 0L ? cnt : res; } } }