View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  
20  package org.apache.hadoop.hbase.io;
21  
22  import java.io.IOException;
23  import java.io.OutputStream;
24  import java.nio.BufferOverflowException;
25  import java.nio.ByteBuffer;
26  import java.nio.ByteOrder;
27  import java.nio.channels.Channels;
28  import java.nio.channels.WritableByteChannel;
29  
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.hbase.classification.InterfaceStability;
32  import org.apache.hadoop.hbase.util.ByteBufferUtils;
33  import org.apache.hadoop.hbase.util.Bytes;
34  
35  /**
36   * Not thread safe!
37   */
38  @InterfaceAudience.Public
39  @InterfaceStability.Evolving
40  public class ByteBufferOutputStream extends OutputStream
41      implements ByteBufferSupportOutputStream {
42    
43    // Borrowed from openJDK:
44    // http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8-b132/java/util/ArrayList.java#221
45    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
46  
47    protected ByteBuffer curBuf = null;
48  
49    ByteBufferOutputStream() {
50
51    }
52  
53    public ByteBufferOutputStream(int capacity) {
54      this(capacity, false);
55    }
56  
57    public ByteBufferOutputStream(int capacity, boolean useDirectByteBuffer) {
58      this(allocate(capacity, useDirectByteBuffer));
59    }
60
61    /**
62     * @param bb ByteBuffer to use. If too small, will be discarded and a new one allocated in its
63     * place; i.e. the passed in BB may NOT BE RETURNED!! Minimally it will be altered. SIDE EFFECT!!
64     * If you want to get the newly allocated ByteBuffer, you'll need to pick it up when
65     * done with this instance by calling {@link #getByteBuffer()}. All this encapsulation violation
66     * is so we can recycle buffers rather than allocate each time; it can get expensive especially
67     * if the buffers are big doing allocations each time or having them undergo resizing because
68     * initial allocation was small.
69     * @see #getByteBuffer()
70     */
71    public ByteBufferOutputStream(final ByteBuffer bb) {
72      assert bb.order() == ByteOrder.BIG_ENDIAN;
73      this.curBuf = bb;
74      this.curBuf.clear();
75    }
76  
77    public int size() {
78      return curBuf.position();
79    }
80
81    private static ByteBuffer allocate(final int capacity, final boolean useDirectByteBuffer) {
82      if (capacity > MAX_ARRAY_SIZE) { // avoid OutOfMemoryError
83        throw new BufferOverflowException();
84      }
85      return useDirectByteBuffer? ByteBuffer.allocateDirect(capacity): ByteBuffer.allocate(capacity);
86    }
87
88    /**
89     * This flips the underlying BB so be sure to use it _last_!
90     * @return ByteBuffer
91     */
92    public ByteBuffer getByteBuffer() {
93      curBuf.flip();
94      return curBuf;
95    }
96
97    protected void checkSizeAndGrow(int extra) {
98      long capacityNeeded = curBuf.position() + (long) extra;
99      if (capacityNeeded > curBuf.limit()) {
100       // guarantee it's possible to fit
101       if (capacityNeeded > MAX_ARRAY_SIZE) {
102         throw new BufferOverflowException();
103       }
104       // double until hit the cap
105       long nextCapacity = Math.min(curBuf.capacity() * 2L, MAX_ARRAY_SIZE);
106       // but make sure there is enough if twice the existing capacity is still too small
107       nextCapacity = Math.max(nextCapacity, capacityNeeded);
108       ByteBuffer newBuf = allocate((int) nextCapacity, curBuf.isDirect());
109       curBuf.flip();
110       ByteBufferUtils.copyFromBufferToBuffer(curBuf, newBuf);
111       curBuf = newBuf;
112     }
113   }
114
115   // OutputStream
116   @Override
117   public void write(int b) throws IOException {
118     checkSizeAndGrow(Bytes.SIZEOF_BYTE);
119     curBuf.put((byte)b);
120   }
121
122  /**
123   * Writes the complete contents of this byte buffer output stream to
124   * the specified output stream argument.
125   *
126   * @param      out   the output stream to which to write the data.
127   * @exception  IOException  if an I/O error occurs.
128   */
129   public void writeTo(OutputStream out) throws IOException {
130     WritableByteChannel channel = Channels.newChannel(out);
131     ByteBuffer bb = curBuf.duplicate();
132     bb.flip();
133     channel.write(bb);
134   }
135
136   @Override
137   public void write(byte[] b) throws IOException {
138     write(b, 0, b.length);
139   }
140
141   @Override
142   public void write(byte[] b, int off, int len) throws IOException {
143     checkSizeAndGrow(len);
144     ByteBufferUtils.copyFromArrayToBuffer(curBuf, b, off, len);
145   }
146
147   public void write(ByteBuffer b, int off, int len) throws IOException {
148     checkSizeAndGrow(len);
149     ByteBufferUtils.copyFromBufferToBuffer(b, curBuf, off, len);
150   }
151
152   /**
153    * Writes an <code>int</code> to the underlying output stream as four
154    * bytes, high byte first.
155    * @param i the <code>int</code> to write
156    * @throws IOException if an I/O error occurs.
157    */
158   public void writeInt(int i) throws IOException {
159     checkSizeAndGrow(Bytes.SIZEOF_INT);
160     ByteBufferUtils.putInt(this.curBuf, i);
161   }
162
163   @Override
164   public void flush() throws IOException {
165     // noop
166   }
167
168   @Override
169   public void close() throws IOException {
170     // noop again. heh
171   }
172 
173   public byte[] toByteArray(int offset, int length) {
174     ByteBuffer bb = curBuf.duplicate();
175     bb.flip();
176
177     byte[] chunk = new byte[length];
178
179     bb.position(offset);
180     bb.get(chunk, 0, length);
181     return chunk;
182   }
183 }