    001/*
    002 * Licensed to the Apache Software Foundation (ASF) under one or more
    003 * contributor license agreements.  See the NOTICE file distributed with
    004 * this work for additional information regarding copyright ownership.
    005 * The ASF licenses this file to You under the Apache License, Version 2.0
    006 * (the "License"); you may not use this file except in compliance with
    007 * the License.  You may obtain a copy of the License at
    008 *
    009 *      http://www.apache.org/licenses/LICENSE-2.0
    010 *
    011 * Unless required by applicable law or agreed to in writing, software
    012 * distributed under the License is distributed on an "AS IS" BASIS,
    013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    014 * See the License for the specific language governing permissions and
    015 * limitations under the License.
    016 */
    017package org.apache.commons.io.output;
    018
    019import java.io.IOException;
    020import java.io.InterruptedIOException;
    021import java.io.OutputStream;
    022import java.io.PipedInputStream;
    023import java.io.PipedOutputStream;
    024import java.util.Objects;
    025import java.util.concurrent.BlockingQueue;
    026import java.util.concurrent.LinkedBlockingQueue;
    027
    028import org.apache.commons.io.input.QueueInputStream;
    029
    030/**
    031 * Simple alternative to JDK {@link PipedOutputStream}; queue input stream provides what's written in queue
    032 * output stream.
    033 * <p>
    034 * Example usage:
    035 * </p>
    036 *
    037 * <pre>
    038 * QueueOutputStream outputStream = new QueueOutputStream();
    039 * QueueInputStream inputStream = outputStream.newPipeInputStream();
    040 *
    041 * outputStream.write("hello world".getBytes(UTF_8));
    042 * inputStream.read();
    043 * </pre>
    044 *
    045 * Unlike JDK {@link PipedInputStream} and {@link PipedOutputStream}, queue input/output streams may be used safely in a
    046 * single thread or multiple threads. Also, unlike JDK classes, no special meaning is attached to initial or current
    047 * thread. Instances can be used longer after initial threads exited.
    048 * <p>
    049 * Closing a {@link QueueOutputStream} has no effect. The methods in this class can be called after the stream has been
    050 * closed without generating an {@link IOException}.
    051 * </p>
    052 *
    053 * @see QueueInputStream
    054 * @since 2.9.0
    055 */
    056public class QueueOutputStream extends OutputStream {
    057
    058    private final BlockingQueue<Integer> blockingQueue;
    059
    060    /**
    061     * Constructs a new instance with no limit to internal buffer size.
    062     */
    063    public QueueOutputStream() {
    064        this(new LinkedBlockingQueue<>());
    065    }
    066
    067    /**
    068     * Constructs a new instance with given buffer.
    069     *
    070     * @param blockingQueue backing queue for the stream
    071     */
    072    public QueueOutputStream(final BlockingQueue<Integer> blockingQueue) {
    073        this.blockingQueue = Objects.requireNonNull(blockingQueue, "blockingQueue");
    074    }
    075
    076    /**
    077     * Constructs a new QueueInputStream instance connected to this. Writes to this output stream will be visible to the
    078     * input stream.
    079     *
    080     * @return QueueInputStream connected to this stream
    081     */
    082    public QueueInputStream newQueueInputStream() {
    083        return QueueInputStream.builder().setBlockingQueue(blockingQueue).get();
    084    }
    085
    086    /**
    087     * Writes a single byte.
    088     *
    089     * @throws InterruptedIOException if the thread is interrupted while writing to the queue.
    090     */
    091    @Override
    092    public void write(final int b) throws InterruptedIOException {
    093        try {
    094            blockingQueue.put(0xFF & b);
    095        } catch (final InterruptedException e) {
    096            Thread.currentThread().interrupt();
    097            final InterruptedIOException interruptedIoException = new InterruptedIOException();
    098            interruptedIoException.initCause(e);
    099            throw interruptedIoException;
    100        }
    101    }
    102}