GCS based Storage Provider for mime4j

GcsStorageProvider.java

import com.google.appengine.tools.cloudstorage.GcsFileOptions;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.GcsInputChannel;
import com.google.appengine.tools.cloudstorage.GcsOutputChannel;
import com.google.appengine.tools.cloudstorage.GcsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.james.mime4j.storage.AbstractStorageProvider;
import org.apache.james.mime4j.storage.Storage;
import org.apache.james.mime4j.storage.StorageOutputStream;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.channels.Channels;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.UUID;

/**
 * A {@link org.apache.james.mime4j.storage.StorageProvider} that stores the data in google cloud storage files. The files
 * are stored in a user specified bucket. User of this class needs to supply the google cloud storage service and bucket name.
 *
 * This implementation is based on {@link org.apache.james.mime4j.storage.TempFileStorageProvider}
 * <p>
 * Example usage:
 *
 * <pre>
 * final String bucketName = "my-bucket";
 * DefaultStorageProvider.setInstance(new GcsStorageProvider(gcsService, bucketName));
 * </pre>
 */
@Slf4j
public class GcsStorageProvider extends AbstractStorageProvider {

	private static final int FETCH_SIZE_MB = 4 * 1024 * 1024;

	private static final String PUBLIC_READ = "public-read";

	private static final GcsFileOptions gcsFileOpts = new GcsFileOptions.Builder().acl(PUBLIC_READ).mimeType("text/csv").build();

	private final GcsService gcsService;

	private final String bucketName;

	/**
	 * Creates a new <code>GcsStorageProvider</code> using the given
	 * values.
	 *
	 * @param gcsService an implementation of {@link GcsService}
	 * @param bucketName google cloud storage bucket name to use.
	 */
	public GcsStorageProvider(final GcsService gcsService, final String bucketName) {
		this.gcsService = gcsService;
		this.bucketName = bucketName;
	}

	@Override
	public StorageOutputStream createStorageOutputStream() throws IOException {
		return new GcsStorageProvider.GcsStorageOutputStream(gcsService, bucketName);
	}

	private static final class GcsStorage implements Storage {

		private final GcsService gcsService;

		private GcsFilename gcsFilename;

		private static final Set<GcsFilename> filesToDelete = new HashSet();

		public GcsStorage(final GcsService gcsService, final GcsFilename gcsFilename) {
			this.gcsService = gcsService;
			this.gcsFilename = gcsFilename;
		}

		@Override
		public InputStream getInputStream() throws IOException {
			if (this.gcsFilename == null) {
				throw new IllegalStateException("storage has been deleted");
			} else {
				final GcsInputChannel readChannel = gcsService.openPrefetchingReadChannel(gcsFilename, 0, FETCH_SIZE_MB);
				return Channels.newInputStream(readChannel);
			}
		}

		@Override
		public void delete() {
			synchronized(filesToDelete) {
				if (this.gcsFilename != null) {
					filesToDelete.add(this.gcsFilename);
					this.gcsFilename = null;
				}

				final Iterator iterator = filesToDelete.iterator();

				while(iterator.hasNext()) {
					GcsFilename filename = (GcsFilename)iterator.next();
					try {
						if (gcsService.delete(filename)) {
                            iterator.remove();
                        }
					} catch (final IOException ex) {
						log.error(ex.getMessage(), ex);
					}
				}

			}
		}
	}

	private static final class GcsStorageOutputStream extends StorageOutputStream {

		private final GcsService gcsService;

		private GcsFilename gcsFilename;

		private final OutputStream outputStream;

		public GcsStorageOutputStream(final GcsService gcsService, final String bucketName) throws IOException {
			this.gcsService = gcsService;

			final String fileName = UUID.randomUUID().toString();
			this.gcsFilename = new GcsFilename(bucketName, fileName);

			GcsOutputChannel gcsOutputChannel = gcsService.createOrReplace(gcsFilename, gcsFileOpts);
			this.outputStream = Channels.newOutputStream(gcsOutputChannel);
		}

		@Override
		protected void write0(byte[] buffer, int offset, int length) throws IOException {
			this.outputStream.write(buffer, offset, length);
		}

		@Override
		protected Storage toStorage0() throws IOException {
			return new GcsStorage(gcsService, gcsFilename);
		}

		@Override
		public void close() throws IOException {
			super.close();
			this.outputStream.close();
		}
	}
}