Author: veithen
Date: Sat May 16 16:57:36 2015
New Revision: 1679747
URL: http://svn.apache.org/r1679747
Log:
Rewrite the MemoryBlob implementation:
- It now allocates buffers with increasing sizes. This should make the
implementation more efficient over a wider range of blob sizes.
- Add a readOnce() method that consumes the content of the blob and gradually
frees the buffers.
Added:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java
(with props)
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java
(with props)
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java
(with props)
Modified:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlob.java
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobImpl.java
Modified:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java?rev=1679747&r1=1679746&r2=1679747&view=diff
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
(original)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/attachments/PartImpl.java
Sat May 16 16:57:36 2015
@@ -20,6 +20,7 @@
package org.apache.axiom.attachments;
import org.apache.axiom.attachments.Part;
+import org.apache.axiom.blob.MemoryBlob;
import org.apache.axiom.blob.OverflowableBlob;
import org.apache.axiom.blob.WritableBlob;
import org.apache.axiom.blob.WritableBlobFactory;
@@ -234,11 +235,13 @@ final class PartImpl implements Part {
return detachableInputStream;
} else {
WritableBlob content = getContent();
- InputStream stream = content.getInputStream();
- if (!preserve) {
- stream = new ReadOnceInputStreamWrapper(this, stream);
+ if (preserve) {
+ return content.getInputStream();
+ } else if (content instanceof MemoryBlob) {
+ return ((MemoryBlob)content).readOnce();
+ } else {
+ return new ReadOnceInputStreamWrapper(this,
content.getInputStream());
}
- return stream;
}
}
Modified:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlob.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlob.java?rev=1679747&r1=1679746&r2=1679747&view=diff
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlob.java
(original)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlob.java
Sat May 16 16:57:36 2015
@@ -35,4 +35,12 @@ public interface MemoryBlob extends Writ
OutputStream getOutputStream();
long getSize();
void release();
+
+ /**
+ * Get an input stream that consumes the content of this blob. The memory
held by this blob will
+ * be gradually released as data is read from the stream.
+ *
+ * @return the input stream to read the data from
+ */
+ InputStream readOnce();
}
Added:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java?rev=1679747&view=auto
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java
(added)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java
Sat May 16 16:57:36 2015
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.axiom.blob;
+
+final class MemoryBlobChunk {
+ final byte[] buffer;
+ int size;
+ MemoryBlobChunk nextChunk;
+
+ MemoryBlobChunk(int capacity) {
+ buffer = new byte[capacity];
+ }
+
+ MemoryBlobChunk allocateNextChunk() {
+ return nextChunk = new MemoryBlobChunk(buffer.length * 2);
+ }
+}
Propchange:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobChunk.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobImpl.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobImpl.java?rev=1679747&r1=1679746&r2=1679747&view=diff
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobImpl.java
(original)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobImpl.java
Sat May 16 16:57:36 2015
@@ -16,242 +16,94 @@
* specific language governing permissions and limitations
* under the License.
*/
-
package org.apache.axiom.blob;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.axiom.ext.io.ReadFromSupport;
import org.apache.axiom.ext.io.StreamCopyException;
final class MemoryBlobImpl implements MemoryBlob {
- // TODO: this should be configurable
- final static int BUFFER_SIZE = 4096;
-
- class OutputStreamImpl extends OutputStream implements ReadFromSupport {
- public void write(byte[] b, int off, int len) throws IOException {
- if (committed) {
- throw new IllegalStateException();
- }
- int total = 0;
- while (total < len) {
- int copy = Math.min(len-total, BUFFER_SIZE-index);
- System.arraycopy(b, off, currBuffer, index, copy);
- total += copy;
- index += copy;
- off += copy;
- if (index >= BUFFER_SIZE) {
- addBuffer();
- }
- }
- }
-
- public void write(byte[] b) throws IOException {
- this.write(b, 0, b.length);
- }
-
- byte[] writeByte = new byte[1];
- public void write(int b) throws IOException {
- writeByte[0] = (byte) b;
- this.write(writeByte, 0, 1);
- }
-
- public void close() throws IOException {
- committed = true;
- }
-
- public long readFrom(InputStream in, long length) throws
StreamCopyException {
- return MemoryBlobImpl.this.readFrom(in, length, false);
- }
- }
-
- class InputStreamImpl extends InputStream {
- private int i;
- private int currIndex;
- private int totalIndex;
- private int mark;
- private byte[] currBuffer;
- private byte[] read_byte = new byte[1];
-
- public InputStreamImpl() {
- }
-
- public int read() throws IOException {
- int read = read(read_byte);
-
- if (read < 0) {
- return -1;
- } else {
- return read_byte[0] & 0xFF;
- }
- }
-
- public int available() throws IOException {
- return (int)getSize() - totalIndex;
- }
-
-
- public synchronized void mark(int readlimit) {
- mark = totalIndex;
- }
-
- public boolean markSupported() {
- return true;
- }
-
- public int read(byte[] b, int off, int len) throws IOException {
- int size = (int)getSize();
- int total = 0;
- if (totalIndex >= size) {
- return -1;
- }
- while (total < len && totalIndex < size) {
- if (currBuffer == null) {
- currBuffer = (byte[]) data.get(i);
- }
- int copy = Math.min(len - total, BUFFER_SIZE - currIndex);
- copy = Math.min(copy, size - totalIndex);
- System.arraycopy(currBuffer, currIndex, b, off, copy);
- total += copy;
- currIndex += copy;
- totalIndex += copy;
- off += copy;
- if (currIndex == BUFFER_SIZE) {
- i++;
- currIndex = 0;
- currBuffer = null;
- }
- }
- return total;
- }
-
- public int read(byte[] b) throws IOException {
- return this.read(b, 0, b.length);
- }
-
- public synchronized void reset() throws IOException {
- i = mark / BUFFER_SIZE;
- currIndex = mark - (i * BUFFER_SIZE);
- currBuffer = (byte[]) data.get(i);
- totalIndex = mark;
- }
- }
-
- List data; // null here indicates the blob is in state NEW
- int index;
- byte[] currBuffer;
- boolean committed;
-
- private void init() {
- data = new ArrayList();
- addBuffer();
- }
-
- void addBuffer() {
- currBuffer = new byte[BUFFER_SIZE];
- data.add(currBuffer);
- index = 0;
- }
+ private MemoryBlobChunk firstChunk;
+ private boolean committed;
public long getSize() {
- if (data == null || !committed) {
+ if (firstChunk == null || !committed) {
throw new IllegalStateException();
} else {
- return (BUFFER_SIZE * (data.size()-1)) + index;
+ long size = 0;
+ MemoryBlobChunk chunk = firstChunk;
+ while (chunk != null) {
+ size += chunk.size;
+ chunk = chunk.nextChunk;
+ }
+ return size;
}
}
public OutputStream getOutputStream() {
- if (data != null || committed) {
+ return internalGetOutputStream();
+ }
+
+ private MemoryBlobOutputStream internalGetOutputStream() {
+ if (firstChunk != null || committed) {
throw new IllegalStateException();
} else {
- init();
- return new OutputStreamImpl();
+ return new MemoryBlobOutputStream(this, firstChunk = new
MemoryBlobChunk(4096));
}
}
- long readFrom(InputStream in, long length, boolean commit) throws
StreamCopyException {
- if (committed) {
- throw new IllegalStateException();
- }
- if (data == null) {
- init();
- }
-
- if (length == -1) {
- length = Long.MAX_VALUE;
- }
- long bytesReceived = 0;
-
- // Now directly write to the buffers
- boolean done = false;
- while (!done) {
-
- // Don't get more than will fit in the current buffer
- int len = (int) Math.min(BUFFER_SIZE - index,
length-bytesReceived);
-
- // Now get the bytes
- int bytesRead;
- try {
- bytesRead = in.read(currBuffer, index, len);
- } catch (IOException ex) {
- throw new StreamCopyException(StreamCopyException.READ, ex);
- }
- if (bytesRead >= 0) {
- bytesReceived += bytesRead;
- index += bytesRead;
- if (index >= BUFFER_SIZE) {
- addBuffer();
- }
- if (bytesReceived >= length) {
- done = true;
- }
- } else {
- done = true;
- }
- }
-
- committed = commit;
-
- return bytesReceived;
+ void commit() {
+ committed = true;
}
-
+
public long readFrom(InputStream in) throws StreamCopyException {
- if (data != null) {
- throw new IllegalStateException();
+ MemoryBlobOutputStream out = internalGetOutputStream();
+ try {
+ return out.readFrom(in, -1);
+ } finally {
+ out.close();
}
- return readFrom(in, -1, true);
}
public InputStream getInputStream() {
- if (data == null || !committed) {
+ return getInputStream(true);
+ }
+
+ public InputStream readOnce() {
+ return getInputStream(false);
+ }
+
+ public InputStream getInputStream(boolean preserve) {
+ if (firstChunk == null || !committed) {
throw new IllegalStateException();
}
- return new InputStreamImpl();
+ InputStream in = new MemoryBlobInputStream(firstChunk);
+ if (!preserve) {
+ firstChunk = null;
+ }
+ return in;
}
public void writeTo(OutputStream os) throws StreamCopyException {
- if (data == null || !committed) {
+ if (firstChunk == null || !committed) {
throw new IllegalStateException();
}
- int size = (int)getSize();
+ MemoryBlobChunk chunk = firstChunk;
try {
- int numBuffers = data.size();
- for (int j = 0; j < numBuffers-1; j ++) {
- os.write( (byte[]) data.get(j), 0, BUFFER_SIZE);
+ while (chunk != null) {
+ if (chunk.size > 0) {
+ os.write(chunk.buffer, 0, chunk.size);
+ }
+ chunk = chunk.nextChunk;
}
- int writeLimit = size - ((numBuffers-1) * BUFFER_SIZE);
- os.write( (byte[]) data.get(numBuffers-1), 0, writeLimit);
} catch (IOException ex) {
throw new StreamCopyException(StreamCopyException.WRITE, ex);
}
}
public void release() {
- data = null;
+ firstChunk = null;
}
}
Added:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java?rev=1679747&view=auto
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java
(added)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java
Sat May 16 16:57:36 2015
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.axiom.blob;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+final class MemoryBlobInputStream extends InputStream {
+ private MemoryBlobChunk chunk;
+ private int index;
+ private MemoryBlobChunk markChunk;
+ private int markIndex;
+
+ MemoryBlobInputStream(MemoryBlobChunk firstChunk) {
+ chunk = firstChunk;
+ }
+
+ private void updateChunk() {
+ while (chunk != null && index == chunk.size) {
+ chunk = chunk.nextChunk;
+ index = 0;
+ }
+ }
+
+ @Override
+ public int read(byte[] buffer, int off, int len) throws IOException {
+ int read = 0;
+ while (len > 0) {
+ updateChunk();
+ if (chunk == null) {
+ if (read == 0) {
+ return -1;
+ } else {
+ break;
+ }
+ }
+ int c = Math.min(len, chunk.size-index);
+ System.arraycopy(chunk.buffer, index, buffer, off, c);
+ index += c;
+ off += c;
+ len -= c;
+ read += c;
+ }
+ return read;
+ }
+
+ @Override
+ public int read() throws IOException {
+ updateChunk();
+ if (chunk == null) {
+ return -1;
+ } else {
+ return chunk.buffer[index++] & 0xFF;
+ }
+ }
+
+ @Override
+ public boolean markSupported() {
+ return true;
+ }
+
+ @Override
+ public synchronized void mark(int readlimit) {
+ markChunk = chunk;
+ markIndex = index;
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ chunk = markChunk;
+ index = markIndex;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ long skipped = 0;
+ while (n > 0) {
+ updateChunk();
+ if (chunk == null) {
+ break;
+ }
+ int c = (int)Math.min(n, chunk.size-index);
+ index += c;
+ skipped += c;
+ n -= c;
+ }
+ return skipped;
+ }
+
+ @Override
+ public int available() throws IOException {
+ if (chunk == null) {
+ return 0;
+ } else {
+ long available = chunk.size - index;
+ MemoryBlobChunk chunk = this.chunk.nextChunk;
+ while (chunk != null) {
+ available += chunk.size;
+ if (available > Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ }
+ chunk = chunk.nextChunk;
+ }
+ return (int)available;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ chunk = null;
+ }
+}
Propchange:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobInputStream.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java
URL:
http://svn.apache.org/viewvc/webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java?rev=1679747&view=auto
==============================================================================
---
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java
(added)
+++
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java
Sat May 16 16:57:36 2015
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.axiom.blob;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.axiom.ext.io.ReadFromSupport;
+import org.apache.axiom.ext.io.StreamCopyException;
+
+final class MemoryBlobOutputStream extends OutputStream implements
ReadFromSupport {
+ private final MemoryBlobImpl blob;
+ private MemoryBlobChunk chunk;
+
+ MemoryBlobOutputStream(MemoryBlobImpl blob, MemoryBlobChunk firstChunk) {
+ this.blob = blob;
+ chunk = firstChunk;
+ }
+
+ private void updateChunk() {
+ if (chunk.size == chunk.buffer.length) {
+ chunk = chunk.allocateNextChunk();
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ if (chunk == null) {
+ throw new IllegalStateException();
+ }
+ int total = 0;
+ while (total < len) {
+ updateChunk();
+ int c = Math.min(len-total, chunk.buffer.length-chunk.size);
+ System.arraycopy(b, off, chunk.buffer, chunk.size, c);
+ chunk.size += c;
+ total += c;
+ off += c;
+ }
+ }
+
+ @Override
+ public void write(int b) {
+ if (chunk == null) {
+ throw new IllegalStateException();
+ }
+ updateChunk();
+ chunk.buffer[chunk.size++] = (byte)b;
+ }
+
+ public long readFrom(InputStream in, long length) throws
StreamCopyException {
+ if (chunk == null) {
+ throw new IllegalStateException();
+ }
+ long read = 0;
+ long toRead = length == -1 ? Long.MAX_VALUE : length;
+ while (toRead > 0) {
+ updateChunk();
+ int c;
+ try {
+ c = in.read(chunk.buffer, chunk.size,
+ (int)Math.min(toRead, chunk.buffer.length-chunk.size));
+ } catch (IOException ex) {
+ throw new StreamCopyException(StreamCopyException.READ, ex);
+ }
+ if (c == -1) {
+ break;
+ }
+ chunk.size += c;
+ read += c;
+ toRead -= c;
+ }
+ return read;
+ }
+
+ @Override
+ public void close() {
+ blob.commit();
+ chunk = null;
+ }
+}
Propchange:
webservices/axiom/trunk/modules/axiom-api/src/main/java/org/apache/axiom/blob/MemoryBlobOutputStream.java
------------------------------------------------------------------------------
svn:eol-style = native
