Refactor all the Cassandra access code into CassandraBlobStore and delegate

to it instead of having it sprayed into three different classes with
accompanying imports and confusion.

Author: Tom May

plugins/cassandra/src/main/java/org/elasticsearch/cassandra/blobstore/AbstractCassandraBlobContainer.java

@@ -22,45 +22,17 @@
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
-import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
 import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
 
-import org.apache.cassandra.thrift.Cassandra;
-import org.apache.cassandra.thrift.Column;
-import org.apache.cassandra.thrift.ColumnParent;
-import org.apache.cassandra.thrift.ColumnPath;
-import org.apache.cassandra.thrift.ColumnOrSuperColumn;
-import org.apache.cassandra.thrift.ConsistencyLevel;
-import org.apache.cassandra.thrift.Deletion;
-import org.apache.cassandra.thrift.InvalidRequestException;
-import org.apache.cassandra.thrift.Mutation;
-import org.apache.cassandra.thrift.SlicePredicate;
-import org.apache.cassandra.thrift.SliceRange;
-import org.apache.cassandra.thrift.TimedOutException;
-import org.apache.cassandra.thrift.UnavailableException;
-
-import org.apache.thrift.TException;
-
 import javax.annotation.Nullable;
 import java.io.IOException;
-import java.nio.charset.Charset;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 
 /**
  * @author Tom May ([email protected])
  */
 public class AbstractCassandraBlobContainer extends AbstractBlobContainer {
-
-    protected static final Charset utf8 = Charset.forName("UTF-8");
-
-    protected static final String keySpace = "ElasticSearch";
-
     protected final ESLogger logger = Loggers.getLogger(getClass());
 
     protected final String blobPath;
@@ -75,172 +47,22 @@ public AbstractCassandraBlobContainer(BlobPath path, CassandraBlobStore blobStor
     }
 
     @Override public boolean blobExists(String blobName) {
-        logger.debug("TODO blobExists {}", blobKey(blobName));
-        try {
-            Cassandra.Client client =
-                CassandraClientFactory.getCassandraClient();
-            try {
-                return client.get_count(
-                    keySpace,
-                    blobKey(blobName),
-                    new ColumnParent("Blobs"),
-                    ConsistencyLevel.QUORUM) != 0;
-            }
-            finally {
-                CassandraClientFactory.closeCassandraClient(client);
-            }
-        } catch (Exception e) {
-            return false;
-        }
+        return blobStore.blobExists(blobPath, blobName);
     }
 
     @Override public boolean deleteBlob(String blobName) throws IOException {
-        logger.debug("deleteBlob {}", blobKey(blobName));
-        Cassandra.Client client =
-            CassandraClientFactory.getCassandraClient();
-        try {
-            long timestamp = System.currentTimeMillis();
-
-            Map<String, Map<String, List<Mutation>>> mutationMap =
-                new HashMap<String, Map<String, List<Mutation>>>();
-
-           // Delete the blob data from Blobs.
-
-            List<Mutation> blobsMutations = new ArrayList<Mutation>();
-            blobsMutations.add(createDelete(null, timestamp));
-
-            Map<String, List<Mutation>> blobsMutationMap =
-                new HashMap<String, List<Mutation>>();
-            blobsMutationMap.put("Blobs", blobsMutations);
-
-            mutationMap.put(blobKey(blobName), blobsMutationMap);
-
-            // Delete the blobName from BlobNames.
-
-            List<Mutation> blobNamesMutations = new ArrayList<Mutation>();
-            blobNamesMutations.add(createDelete(blobName, timestamp));
-
-            Map<String, List<Mutation>> blobNamesMutationMap =
-                new HashMap<String, List<Mutation>>();
-            blobNamesMutationMap.put("BlobNames", blobNamesMutations);
-
-            mutationMap.put(blobPath, blobNamesMutationMap);
-
-            client.batch_mutate(
-                keySpace, mutationMap, ConsistencyLevel.QUORUM);
-
-            return true;
-        }
-        catch (Exception e) {
-            // TODO S3 does this, what's the deal with returning false
-            // vs. throwing IOException?
-            return false;
-        }
-        finally {
-            CassandraClientFactory.closeCassandraClient(client);
-        }
+        return blobStore.deleteBlob(blobPath, blobName);
     }
 
-    @Override public void readBlob(final String blobName, final ReadBlobListener listener) {
-        logger.debug("readBlob {}", blobKey(blobName));
-        blobStore.executor().execute(new Runnable() {
-            @Override public void run() {
-                Cassandra.Client client = null;
-                try {
-                    client = CassandraClientFactory.getCassandraClient();
-                    readBlob(client, blobName, listener);
-                }
-                catch (Exception ex) {
-                    listener.onFailure(ex);
-                }
-                finally {
-                    if (client != null) {
-                        CassandraClientFactory.closeCassandraClient(client);
-                    }
-                }
-            }
-        });
-    }
-
-    private void readBlob(Cassandra.Client client, String blobName, ReadBlobListener listener)
-        throws Exception
-    {
-        ColumnOrSuperColumn columnOrSuperColumn = client.get(
-            keySpace,
-            blobKey(blobName),
-            new ColumnPath("Blobs").setColumn(utf8.encode("data")),
-            ConsistencyLevel.QUORUM);
-        Column column = columnOrSuperColumn.getColumn();
-        byte[] blobData = column.getValue();
-        logger.debug("Read {} bytes: {}", blobKey(blobName), blobData.length);
-        listener.onPartial(blobData, 0, blobData.length);
-        listener.onCompleted();
+    @Override public void readBlob(String blobName, ReadBlobListener listener) {
+        blobStore.readBlob(blobPath, blobName, listener);
     }
 
     @Override public ImmutableMap<String, BlobMetaData> listBlobsByPrefix(@Nullable String blobNamePrefix) throws IOException {
-        logger.debug("listBlobsByPrefix {}", blobKey(blobNamePrefix));
-
-        List<ColumnOrSuperColumn> columns;
-        Cassandra.Client client = CassandraClientFactory.getCassandraClient();
-        try {
-            columns = client.get_slice(
-                keySpace,
-                blobPath,
-                new ColumnParent("BlobNames"),
-                new SlicePredicate().setSlice_range(
-                    new SliceRange()
-                    .setStart(new byte[0])
-                    .setFinish(new byte[0])
-                    .setCount(1000000000)),
-                ConsistencyLevel.QUORUM);
-        }
-        catch (InvalidRequestException ex) {
-            throw new IOException("Cassandra get_slice on ???:??? failed", ex);
-        }
-        catch (UnavailableException ex) {
-            throw new IOException("Cassandra get_slice on ???:??? failed", ex);
-        }
-        catch (TimedOutException ex) {
-            throw new IOException("Cassandra get_slice on ???:??? failed", ex);
-        }
-        catch (TException ex) {
-            throw new IOException("Cassandra get_slice on ???:??? failed", ex);
-        }
-        finally {
-            CassandraClientFactory.closeCassandraClient(client);
-        }
-
-        ImmutableMap.Builder<String, BlobMetaData> blobsBuilder = ImmutableMap.builder();
-
-        for (ColumnOrSuperColumn columnOrSuperColumn : columns) {
-            Column column = columnOrSuperColumn.getColumn();
-            String name = new String(column.getName(), utf8);
-            long length = Integer.parseInt(new String(column.getValue(), utf8));
-            logger.debug("name: {} length: {}", name, length);
-            if (blobNamePrefix == null || name.startsWith(blobNamePrefix)) {
-                blobsBuilder.put(name, new PlainBlobMetaData(name, length));
-            }
-        }
-
-        return blobsBuilder.build();
+        return blobStore.listBlobsByPrefix(blobPath, blobNamePrefix);
     }
 
     @Override public ImmutableMap<String, BlobMetaData> listBlobs() throws IOException {
         return listBlobsByPrefix(null);
     }
-
-    protected String blobKey(String blobName) {
-        return blobPath + '/' + blobName;
-    }
-
-    private Mutation createDelete(String name, long timestamp) {
-        Deletion deletion = new Deletion(timestamp);
-        if (name != null) {
-            List<ByteBuffer> columnNames = new ArrayList<ByteBuffer>(1);
-            columnNames.add(utf8.encode(name));
-            deletion.setPredicate(
-                new SlicePredicate().setColumn_names(columnNames));
-        }
-        return new Mutation().setDeletion(deletion);
-    }
 }