| /* |
| * Copyright 2001-2004 The Apache Software Foundation. |
| * |
| * 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.apache.axis.attachments; |
| |
| import org.apache.axis.InternalException; |
| import org.apache.axis.MessageContext; |
| import org.apache.axis.components.logger.LogFactory; |
| import org.apache.axis.utils.Messages; |
| import org.apache.commons.logging.Log; |
| |
| import java.io.File; |
| import java.io.BufferedInputStream; |
| |
| /** |
| * This class allows small attachments to be cached in memory, while large ones are |
| * cached out. It implements a Java Activiation Data source interface. |
| * |
| * @author Rick Rineholt |
| */ |
| public class ManagedMemoryDataSource implements javax.activation.DataSource { |
| |
| /** Field log */ |
| protected static Log log = |
| LogFactory.getLog(ManagedMemoryDataSource.class.getName()); |
| |
| /** |
| * The content type. This defaults to |
| * <code>application/octet-stream</code>. |
| */ |
| protected String contentType = "application/octet-stream"; |
| |
| /** The incoming source stream. */ |
| java.io.InputStream ss = null; |
| |
| /** Field MIN_MEMORY_DISK_CACHED */ |
| public static final int MIN_MEMORY_DISK_CACHED = -1; |
| |
| /** Field MAX_MEMORY_DISK_CACHED */ |
| public static final int MAX_MEMORY_DISK_CACHED = 16 * 1024; |
| |
| /** Field maxCached */ |
| protected int maxCached = MAX_MEMORY_DISK_CACHED; // max in memory cached. Default. |
| |
| // If set the file the disk is cached to. |
| |
| /** Field diskCacheFile */ |
| protected java.io.File diskCacheFile = null; |
| |
| // A list of open input Streams. |
| |
| /** Field readers */ |
| protected java.util.WeakHashMap readers = new java.util.WeakHashMap(); |
| |
| /** |
| * Flag to show if the resources behind this have been deleted. |
| */ |
| protected boolean deleted = |
| false; |
| |
| // Memory is allocated in these size chunks. |
| |
| /** Field READ_CHUNK_SZ */ |
| public static final int READ_CHUNK_SZ = 32 * 1024; |
| |
| /** Field debugEnabled */ |
| protected boolean debugEnabled = false; // Log debugging if true. |
| |
| // Should not be called; |
| |
| /** |
| * Constructor ManagedMemoryDataSource. |
| */ |
| protected ManagedMemoryDataSource() { |
| } |
| |
| /** |
| * Create a new boundary stream. |
| * |
| * @param ss is the source input stream that is used to create this data source. |
| * @param maxCached This is the max memory that is to be used to cache the data. |
| * @param contentType the mime type for this data stream. |
| * by buffering you can some effiency in searching. |
| * |
| * @throws java.io.IOException |
| */ |
| public ManagedMemoryDataSource( |
| java.io.InputStream ss, int maxCached, String contentType) |
| throws java.io.IOException { |
| this(ss, maxCached, contentType, false); |
| } |
| |
| /** |
| * Create a new boundary stream. |
| * |
| * @param ss is the source input stream that is used to create this data source. |
| * @param maxCached This is the max memory that is to be used to cache the data. |
| * @param contentType the mime type for this data stream. |
| * by buffering you can some effiency in searching. |
| * @param readall if true will read in the whole source. |
| * |
| * @throws java.io.IOException |
| */ |
| public ManagedMemoryDataSource( |
| java.io.InputStream ss, int maxCached, String contentType, boolean readall) |
| throws java.io.IOException { |
| |
| if(ss instanceof BufferedInputStream) { |
| this.ss = ss; |
| } else { |
| this.ss = new BufferedInputStream(ss); |
| } |
| this.maxCached = maxCached; |
| |
| if ((null != contentType) && (contentType.length() != 0)) { |
| this.contentType = contentType; |
| } |
| |
| if (maxCached < MIN_MEMORY_DISK_CACHED) { |
| throw new IllegalArgumentException( |
| Messages.getMessage("badMaxCached", "" + maxCached)); |
| } |
| |
| if (log.isDebugEnabled()) { |
| debugEnabled = true; // Logging should be initialized by time; |
| } |
| |
| // for now read all in to disk. |
| if (readall) { |
| byte[] readbuffer = new byte[READ_CHUNK_SZ]; |
| int read = 0; |
| |
| do { |
| read = ss.read(readbuffer); |
| |
| if (read > 0) { |
| write(readbuffer, read); |
| } |
| } while (read > -1); |
| |
| close(); |
| } |
| } |
| |
| /* javax.activation.Interface DataSource implementation */ |
| |
| /** |
| * This method returns the MIME type of the data in the form of a string. |
| * @return The mime type. |
| */ |
| public java.lang.String getContentType() { |
| return contentType; |
| } |
| |
| /** |
| * This method returns an InputStream representing the the data and throws the appropriate exception if it can not do so. |
| * @return the java.io.InputStream for the data source. |
| * |
| * @throws java.io.IOException |
| */ |
| public synchronized java.io.InputStream getInputStream() |
| throws java.io.IOException { |
| |
| /* |
| * if (memorybuflist == null) { |
| * return new java.io.FileInputStream(diskCacheFile); |
| * } |
| * else |
| */ |
| return new Instream(); // Return the memory held stream. |
| } |
| |
| /** |
| * This will flush any memory source to disk and |
| * provide the name of the file if desired. |
| * |
| * @return the name of the file of the stream |
| */ |
| public java.lang.String getName() { |
| |
| String ret = null; |
| |
| try { |
| flushToDisk(); |
| |
| if (diskCacheFile != null) { |
| ret = diskCacheFile.getAbsolutePath(); |
| } |
| } catch (Exception e) { |
| diskCacheFile = null; |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * This method returns an OutputStream where the data can be written and |
| * throws the appropriate exception if it can not do so. |
| * NOT SUPPORTED, not need for axis, data sources are create by constructors. |
| * |
| * |
| * @return always <code>null</code> |
| * |
| * @throws java.io.IOException |
| */ |
| public java.io.OutputStream getOutputStream() throws java.io.IOException { |
| return null; |
| } |
| |
| /** The linked list to hold the in memory buffers. */ |
| protected java.util.LinkedList memorybuflist = |
| new java.util.LinkedList(); |
| |
| /** Hold the last memory buffer. */ |
| protected byte[] currentMemoryBuf = null; |
| |
| /** The number of bytes written to the above buffer. */ |
| protected int currentMemoryBufSz = |
| 0; |
| |
| /** The total size in bytes in this data source. */ |
| protected long totalsz = 0; |
| |
| /** This is the cached disk stream. */ |
| protected java.io.BufferedOutputStream cachediskstream = |
| null; |
| |
| /** If true the source input stream is now closed. */ |
| protected boolean closed = false; |
| |
| /** |
| * Write bytes to the stream. |
| * |
| * @param data all bytes of this array are written to the stream |
| * @throws java.io.IOException if there was a problem writing the data |
| */ |
| protected void write(byte[] data) throws java.io.IOException { |
| write(data, data.length); |
| } |
| |
| /** |
| * This method is a low level write. |
| * Note it is designed to in the future to allow streaming to both memory |
| * AND to disk simultaneously. |
| * |
| * @param data |
| * @param length |
| * |
| * @throws java.io.IOException |
| */ |
| protected synchronized void write(byte[] data, int length) |
| throws java.io.IOException { |
| |
| if (closed) { |
| throw new java.io.IOException(Messages.getMessage("streamClosed")); |
| } |
| |
| int writesz = length; |
| int byteswritten = 0; |
| |
| if ((null != memorybuflist) |
| && (totalsz + writesz > maxCached)) { // Cache to disk. |
| if (null == cachediskstream) { // Need to create a disk cache |
| flushToDisk(); |
| } |
| } |
| |
| if (memorybuflist != null) { // Can write to memory. |
| do { |
| if (null == currentMemoryBuf) { |
| currentMemoryBuf = new byte[READ_CHUNK_SZ]; |
| currentMemoryBufSz = 0; |
| |
| memorybuflist.add(currentMemoryBuf); |
| } |
| |
| // bytes to write is the min. between the remaining bytes and what is left in this buffer. |
| int bytes2write = Math.min((writesz - byteswritten), |
| (currentMemoryBuf.length |
| - currentMemoryBufSz)); |
| |
| // copy the data. |
| System.arraycopy(data, byteswritten, currentMemoryBuf, |
| currentMemoryBufSz, bytes2write); |
| |
| byteswritten += bytes2write; |
| currentMemoryBufSz += bytes2write; |
| |
| if (byteswritten |
| < writesz) { // only get more if we really need it. |
| currentMemoryBuf = new byte[READ_CHUNK_SZ]; |
| currentMemoryBufSz = 0; |
| |
| memorybuflist.add(currentMemoryBuf); // add it to the chain. |
| } |
| } while (byteswritten < writesz); |
| } |
| |
| if (null != cachediskstream) { // Write to the out going stream. |
| cachediskstream.write(data, 0, length); |
| } |
| |
| totalsz += writesz; |
| |
| return; |
| } |
| |
| /** |
| * This method is a low level write. |
| * Close the stream. |
| * |
| * @throws java.io.IOException |
| */ |
| protected synchronized void close() throws java.io.IOException { |
| |
| if (!closed) { |
| closed = true; // Markit as closed. |
| |
| if (null != cachediskstream) { // close the disk cache. |
| cachediskstream.close(); |
| |
| cachediskstream = null; |
| } |
| |
| if (null != memorybuflist) { // There is a memory buffer. |
| if (currentMemoryBufSz > 0) { |
| byte[] tmp = |
| new byte[currentMemoryBufSz]; // Get the last buffer and make it the sizeof the actual data. |
| |
| System.arraycopy(currentMemoryBuf, 0, tmp, 0, |
| currentMemoryBufSz); |
| memorybuflist.set( |
| memorybuflist.size() - 1, |
| tmp); // Now replace the last buffer with this size. |
| } |
| |
| currentMemoryBuf = null; // No need for this anymore. |
| } |
| } |
| } |
| |
| protected void finalize() throws Throwable { |
| |
| if (null != cachediskstream) { // close the disk cache. |
| cachediskstream.close(); |
| |
| cachediskstream = null; |
| } |
| } |
| |
| /** |
| * Routine to flush data to disk if is in memory. |
| * |
| * @throws java.io.IOException |
| * @throws java.io.FileNotFoundException |
| */ |
| protected void flushToDisk() |
| throws java.io.IOException, java.io.FileNotFoundException { |
| |
| java.util.LinkedList ml = memorybuflist; |
| |
| log.debug(Messages.getMessage("maxCached", "" + maxCached, |
| "" + totalsz)); |
| |
| if (ml != null) { |
| if (null == cachediskstream) { // Need to create a disk cache |
| try { |
| MessageContext mc = MessageContext.getCurrentContext(); |
| String attdir = (mc == null) |
| ? null |
| : mc.getStrProp( |
| MessageContext.ATTACHMENTS_DIR); |
| |
| diskCacheFile = java.io.File.createTempFile("Axis", ".att", |
| (attdir == null) |
| ? null |
| : new File( |
| attdir)); |
| |
| if(log.isDebugEnabled()) { |
| log.debug( |
| Messages.getMessage( |
| "diskCache", diskCacheFile.getAbsolutePath())); |
| } |
| |
| cachediskstream = new java.io.BufferedOutputStream( |
| new java.io.FileOutputStream(diskCacheFile)); |
| |
| int listsz = ml.size(); |
| |
| // Write out the entire memory held store to disk. |
| for (java.util.Iterator it = ml.iterator(); |
| it.hasNext();) { |
| byte[] rbuf = (byte[]) it.next(); |
| int bwrite = (listsz-- == 0) |
| ? currentMemoryBufSz |
| : rbuf.length; |
| |
| cachediskstream.write(rbuf, 0, bwrite); |
| |
| if (closed) { |
| cachediskstream.close(); |
| |
| cachediskstream = null; |
| } |
| } |
| |
| memorybuflist = null; |
| } catch (java.lang.SecurityException se) { |
| diskCacheFile = null; |
| cachediskstream = null; |
| maxCached = java.lang.Integer.MAX_VALUE; |
| |
| log.info(Messages.getMessage("nodisk00"), se); |
| } |
| } |
| } |
| } |
| |
| public synchronized boolean delete() { |
| |
| boolean ret = false; |
| |
| deleted = true; |
| |
| memorybuflist = null; |
| |
| if (diskCacheFile != null) { |
| if (cachediskstream != null) { |
| try { |
| cachediskstream.close(); |
| } catch (Exception e) { |
| } |
| |
| cachediskstream = null; |
| } |
| |
| Object[] array = readers.keySet().toArray(); |
| for (int i = 0; i < array.length; i++) { |
| Instream stream = (Instream) array[i]; |
| if (null != stream) { |
| try { |
| stream.close(); |
| } catch (Exception e) { |
| } |
| } |
| } |
| readers.clear(); |
| |
| try { |
| diskCacheFile.delete(); |
| |
| ret = true; |
| } catch (Exception e) { |
| |
| // Give it our best shot. |
| diskCacheFile.deleteOnExit(); |
| } |
| } |
| |
| |
| return ret; |
| } |
| |
| // inner classes cannot have static declarations... |
| |
| /** Field is_log */ |
| protected static Log is_log = |
| LogFactory.getLog(Instream.class.getName()); |
| |
| /** |
| * Inner class to handle getting an input stream to this data source |
| * Handles creating an input stream to the source. |
| */ |
| private class Instream extends java.io.InputStream { |
| |
| /** bytes read. */ |
| protected long bread = 0; |
| |
| /** The real stream. */ |
| java.io.FileInputStream fin = null; |
| |
| /** The position in the list were we are reading from. */ |
| int currentIndex = |
| 0; |
| |
| /** the buffer we are currently reading from. */ |
| byte[] currentBuf = null; |
| |
| /** The current position in there. */ |
| int currentBufPos = 0; |
| |
| /** The read stream has been closed. */ |
| boolean readClosed = false; |
| |
| /** |
| * Constructor Instream. |
| * |
| * @throws java.io.IOException if the Instream could not be created or |
| * if the data source has been deleted |
| */ |
| protected Instream() throws java.io.IOException { |
| |
| if (deleted) { |
| throw new java.io.IOException( |
| Messages.getMessage("resourceDeleted")); |
| } |
| |
| readers.put(this, null); |
| } |
| |
| /** |
| * Query for the number of bytes available for reading. |
| * |
| * @return the number of bytes left |
| * |
| * @throws java.io.IOException if this stream is not in a state that |
| * supports reading |
| */ |
| public int available() throws java.io.IOException { |
| |
| if (deleted) { |
| throw new java.io.IOException( |
| Messages.getMessage("resourceDeleted")); |
| } |
| |
| if (readClosed) { |
| throw new java.io.IOException( |
| Messages.getMessage("streamClosed")); |
| } |
| |
| int ret = new Long(Math.min(Integer.MAX_VALUE, totalsz - bread)).intValue(); |
| |
| if (debugEnabled) { |
| is_log.debug("available() = " + ret + "."); |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * Read a byte from the stream. |
| * |
| * @return byte read or -1 if no more data. |
| * |
| * @throws java.io.IOException |
| */ |
| public int read() throws java.io.IOException { |
| |
| synchronized (ManagedMemoryDataSource.this) { |
| byte[] retb = new byte[1]; |
| int br = read(retb, 0, 1); |
| |
| if (br == -1) { |
| return -1; |
| } |
| return 0xFF & retb[0]; |
| } |
| } |
| |
| /** |
| * Not supported. |
| * |
| * @return |
| */ |
| public boolean markSupported() { |
| |
| if (debugEnabled) { |
| is_log.debug("markSupported() = " + false + "."); |
| } |
| |
| return false; |
| } |
| |
| /** |
| * Not supported. |
| * |
| * @param readlimit |
| */ |
| public void mark(int readlimit) { |
| |
| if (debugEnabled) { |
| is_log.debug("mark()"); |
| } |
| } |
| |
| /** |
| * Not supported. |
| * |
| * @throws java.io.IOException |
| */ |
| public void reset() throws java.io.IOException { |
| |
| if (debugEnabled) { |
| is_log.debug("reset()"); |
| } |
| |
| throw new java.io.IOException(Messages.getMessage("noResetMark")); |
| } |
| |
| public long skip(long skipped) throws java.io.IOException { |
| |
| if (debugEnabled) { |
| is_log.debug("skip(" + skipped + ")."); |
| } |
| |
| if (deleted) { |
| throw new java.io.IOException( |
| Messages.getMessage("resourceDeleted")); |
| } |
| |
| if (readClosed) { |
| throw new java.io.IOException( |
| Messages.getMessage("streamClosed")); |
| } |
| |
| if (skipped < 1) { |
| return 0; // nothing to skip. |
| } |
| |
| synchronized (ManagedMemoryDataSource.this) { |
| skipped = Math.min(skipped, |
| totalsz |
| - bread); // only skip what we've read. |
| |
| if (skipped == 0) { |
| return 0; |
| } |
| |
| java.util.List ml = memorybuflist; // hold the memory list. |
| int bwritten = 0; |
| |
| if (ml != null) { |
| if (null == currentBuf) { // get the buffer we need to read from. |
| currentBuf = (byte[]) ml.get(currentIndex); |
| currentBufPos = 0; // start reading from the begining. |
| } |
| |
| do { |
| long bcopy = Math.min(currentBuf.length |
| - currentBufPos, |
| skipped - bwritten); |
| |
| bwritten += bcopy; |
| currentBufPos += bcopy; |
| |
| if (bwritten < skipped) { |
| currentBuf = (byte[]) ml.get(++currentIndex); |
| currentBufPos = 0; |
| } |
| } while (bwritten < skipped); |
| } |
| |
| if (null != fin) { |
| fin.skip(skipped); |
| } |
| |
| bread += skipped; |
| } |
| |
| if (debugEnabled) { |
| is_log.debug("skipped " + skipped + "."); |
| } |
| |
| return skipped; |
| } |
| |
| public int read(byte[] b, int off, int len) throws java.io.IOException { |
| |
| if (debugEnabled) { |
| is_log.debug(this.hashCode() + " read(" + off + ", " + len |
| + ")"); |
| } |
| |
| if (deleted) { |
| throw new java.io.IOException( |
| Messages.getMessage("resourceDeleted")); |
| } |
| |
| if (readClosed) { |
| throw new java.io.IOException( |
| Messages.getMessage("streamClosed")); |
| } |
| |
| if (b == null) { |
| throw new InternalException(Messages.getMessage("nullInput")); |
| } |
| |
| if (off < 0) { |
| throw new IndexOutOfBoundsException( |
| Messages.getMessage("negOffset", "" + off)); |
| } |
| |
| if (len < 0) { |
| throw new IndexOutOfBoundsException( |
| Messages.getMessage("length", "" + len)); |
| } |
| |
| if (len + off > b.length) { |
| throw new IndexOutOfBoundsException( |
| Messages.getMessage("writeBeyond")); |
| } |
| |
| if (len == 0) { |
| return 0; |
| } |
| |
| int bwritten = 0; |
| |
| synchronized (ManagedMemoryDataSource.this) { |
| if (bread == totalsz) { |
| return -1; |
| } |
| |
| java.util.List ml = memorybuflist; |
| |
| long longlen = len; |
| longlen = Math.min( |
| longlen, |
| totalsz |
| - bread); // Only return the number of bytes in the data store that is left. |
| len = new Long(longlen).intValue(); |
| |
| if (debugEnabled) { |
| is_log.debug("len = " + len); |
| } |
| |
| if (ml != null) { |
| if (null == currentBuf) { // Get the buffer we need to read from. |
| currentBuf = (byte[]) ml.get(currentIndex); |
| currentBufPos = 0; // New buffer start from the begining. |
| } |
| |
| do { |
| |
| // The bytes to copy, the minimum of the bytes left in this buffer or bytes remaining. |
| int bcopy = Math.min(currentBuf.length - currentBufPos, |
| len - bwritten); |
| |
| // Copy the data. |
| System.arraycopy(currentBuf, currentBufPos, b, |
| off + bwritten, bcopy); |
| |
| bwritten += bcopy; |
| currentBufPos += bcopy; |
| |
| if (bwritten < len) { // Get the next buffer. |
| currentBuf = (byte[]) ml.get(++currentIndex); |
| currentBufPos = 0; |
| } |
| } while (bwritten < len); |
| } |
| |
| if ((bwritten == 0) && (null != diskCacheFile)) { |
| if (debugEnabled) { |
| is_log.debug(Messages.getMessage("reading", "" + len)); |
| } |
| |
| if (null == fin) { // we are now reading from disk. |
| if (debugEnabled) { |
| is_log.debug( |
| Messages.getMessage( |
| "openBread", |
| diskCacheFile.getCanonicalPath())); |
| } |
| |
| if (debugEnabled) { |
| is_log.debug(Messages.getMessage("openBread", |
| "" + bread)); |
| } |
| |
| fin = new java.io.FileInputStream(diskCacheFile); |
| |
| if (bread > 0) { |
| fin.skip(bread); // Skip what we've read so far. |
| } |
| } |
| |
| if (cachediskstream != null) { |
| if (debugEnabled) { |
| is_log.debug(Messages.getMessage("flushing")); |
| } |
| |
| cachediskstream.flush(); |
| } |
| |
| if (debugEnabled) { |
| is_log.debug(Messages.getMessage("flushing")); |
| is_log.debug("len=" + len); |
| is_log.debug("off=" + off); |
| is_log.debug("b.length=" + b.length); |
| } |
| |
| bwritten = fin.read(b, off, len); |
| } |
| |
| if (bwritten > 0) { |
| bread += bwritten; |
| } |
| } |
| |
| if (debugEnabled) { |
| is_log.debug(this.hashCode() |
| + Messages.getMessage("read", "" + bwritten)); |
| } |
| |
| return bwritten; |
| } |
| |
| /** |
| * close the stream. |
| * |
| * @throws java.io.IOException |
| */ |
| public synchronized void close() throws java.io.IOException { |
| |
| if (debugEnabled) { |
| is_log.debug("close()"); |
| } |
| |
| if (!readClosed) { |
| readers.remove(this); |
| |
| readClosed = true; |
| |
| if (fin != null) { |
| fin.close(); |
| } |
| |
| fin = null; |
| } |
| } |
| |
| protected void finalize() throws Throwable { |
| close(); |
| } |
| } // endof innerclass Instream |
| |
| // Used to test. |
| |
| /** |
| * Method main |
| * |
| * @param arg |
| */ |
| public static void main(String arg[]) { // test |
| |
| try { |
| String readFile = arg[0]; |
| String writeFile = arg[1]; |
| java.io.FileInputStream ss = |
| new java.io.FileInputStream(readFile); |
| ManagedMemoryDataSource ms = |
| new ManagedMemoryDataSource(ss, 1024 * 1024, "foo/data", true); |
| javax.activation.DataHandler dh = |
| new javax.activation.DataHandler(ms); |
| java.io.InputStream is = dh.getInputStream(); |
| java.io.FileOutputStream fo = |
| new java.io.FileOutputStream(writeFile); |
| byte[] buf = new byte[512]; |
| int read = 0; |
| |
| do { |
| read = is.read(buf); |
| |
| if (read > 0) { |
| fo.write(buf, 0, read); |
| } |
| } while (read > -1); |
| |
| fo.close(); |
| is.close(); |
| } catch (java.lang.Exception e) { |
| log.error(Messages.getMessage("exception00"), e); |
| } |
| } |
| |
| /** |
| * get the filename of the content if it is cached to disk. |
| * @return file object pointing to file, or null for memory-stored content |
| */ |
| public File getDiskCacheFile() { |
| return diskCacheFile; |
| } |
| } |
