|
| 1 | +package com.datastax.driver.examples.datatypes; |
| 2 | + |
| 3 | +import com.datastax.driver.core.*; |
| 4 | +import com.datastax.driver.core.utils.Bytes; |
| 5 | +import com.google.common.collect.ImmutableMap; |
| 6 | + |
| 7 | +import java.io.*; |
| 8 | +import java.nio.ByteBuffer; |
| 9 | +import java.nio.channels.FileChannel; |
| 10 | +import java.util.Map; |
| 11 | + |
| 12 | +/** |
| 13 | + * Inserts and retrieves values in BLOB columns. |
| 14 | + * <p/> |
| 15 | + * By default, the Java driver maps this type to {@link java.nio.ByteBuffer}. The ByteBuffer API is a bit tricky to use |
| 16 | + * at times, so we will show common pitfalls as well. We strongly recommend that you read the {@link java.nio.Buffer} |
| 17 | + * and {@link ByteBuffer} API docs and become familiar with the capacity, limit and position properties. |
| 18 | + * <a href="http://tutorials.jenkov.com/java-nio/buffers.html">This tutorial</a> might also help. |
| 19 | + * <p/> |
| 20 | + * Preconditions: |
| 21 | + * - a Cassandra cluster is running and accessible through the contacts points identified by CONTACT_POINTS and PORT; |
| 22 | + * - FILE references an existing file. |
| 23 | + * <p/> |
| 24 | + * Side effects: |
| 25 | + * - creates a new keyspace "examples" in the cluster. It a keyspace with this name already exists, it will be reused; |
| 26 | + * - creates a table "examples.blobs". If it already exists, it will be reused; |
| 27 | + * - inserts data in the table. |
| 28 | + */ |
| 29 | +public class Blobs { |
| 30 | + |
| 31 | + static String[] CONTACT_POINTS = {"127.0.0.1"}; |
| 32 | + static int PORT = 9042; |
| 33 | + |
| 34 | + static File FILE = new File(Blobs.class.getResource("/cassandra_logo.png").getFile()); |
| 35 | + |
| 36 | + public static void main(String[] args) throws IOException { |
| 37 | + Cluster cluster = null; |
| 38 | + try { |
| 39 | + cluster = Cluster.builder() |
| 40 | + .addContactPoints(CONTACT_POINTS).withPort(PORT) |
| 41 | + .build(); |
| 42 | + Session session = cluster.connect(); |
| 43 | + |
| 44 | + createSchema(session); |
| 45 | + allocateAndInsert(session); |
| 46 | + retrieveSimpleColumn(session); |
| 47 | + retrieveMapColumn(session); |
| 48 | + insertConcurrent(session); |
| 49 | + insertFromAndRetrieveToFile(session); |
| 50 | + } finally { |
| 51 | + if (cluster != null) cluster.close(); |
| 52 | + } |
| 53 | + } |
| 54 | + |
| 55 | + private static void createSchema(Session session) { |
| 56 | + session.execute("CREATE KEYSPACE IF NOT EXISTS examples " + |
| 57 | + "WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}"); |
| 58 | + session.execute("CREATE TABLE IF NOT EXISTS examples.blobs(k int PRIMARY KEY, b blob, m map<text, blob>)"); |
| 59 | + } |
| 60 | + |
| 61 | + private static void allocateAndInsert(Session session) { |
| 62 | + // One way to get a byte buffer is to allocate it and fill it yourself: |
| 63 | + ByteBuffer buffer = ByteBuffer.allocate(16); |
| 64 | + while (buffer.hasRemaining()) |
| 65 | + buffer.put((byte) 0xFF); |
| 66 | + |
| 67 | + // Don't forget to flip! The driver expects a buffer that is ready for reading. That is, it will consider all |
| 68 | + // the data between buffer.position() and buffer.limit(). |
| 69 | + // Right now we are positioned at the end because we just finished writing, so if we passed the buffer as-is it |
| 70 | + // would appear to be empty: |
| 71 | + assert buffer.limit() - buffer.position() == 0; |
| 72 | + |
| 73 | + buffer.flip(); |
| 74 | + // Now position is back to the beginning, so the driver will see all 16 bytes. |
| 75 | + assert buffer.limit() - buffer.position() == 16; |
| 76 | + |
| 77 | + session.execute("INSERT INTO examples.blobs (k, b, m) VALUES (1, ?, ?)", |
| 78 | + buffer, ImmutableMap.of("test", buffer)); |
| 79 | + } |
| 80 | + |
| 81 | + private static void retrieveSimpleColumn(Session session) { |
| 82 | + Row row = session.execute("SELECT b, m FROM examples.blobs WHERE k = 1").one(); |
| 83 | + |
| 84 | + ByteBuffer buffer = row.getBytes("b"); |
| 85 | + |
| 86 | + // The driver always returns buffers that are ready for reading. |
| 87 | + assert buffer.limit() - buffer.position() == 16; |
| 88 | + |
| 89 | + // One way to read from the buffer is to use absolute getters. Do NOT start reading at index 0, as the buffer |
| 90 | + // might start at a different position (we'll see an example of that later). |
| 91 | + for (int i = buffer.position(); i < buffer.limit(); i++) { |
| 92 | + assert buffer.get(i) == (byte) 0xFF; |
| 93 | + } |
| 94 | + |
| 95 | + // Another way is to use relative getters. |
| 96 | + while (buffer.hasRemaining()) { |
| 97 | + assert buffer.get() == (byte) 0xFF; |
| 98 | + } |
| 99 | + // Note that relative getters change the position, so when we're done reading we're at the end again. |
| 100 | + assert buffer.position() == buffer.limit(); |
| 101 | + |
| 102 | + // Reset the position for the next operation. |
| 103 | + buffer.flip(); |
| 104 | + |
| 105 | + // Yet another way is to convert the buffer to a byte array. Do NOT use buffer.array(), because it returns the |
| 106 | + // buffer's *backing array*, which is not the same thing as its contents: |
| 107 | + // - not all byte buffers have backing arrays |
| 108 | + // - even then, the backing array might be larger than the buffer's contents |
| 109 | + // |
| 110 | + // The driver provides a utility method that handles those details for you: |
| 111 | + byte[] array = Bytes.getArray(buffer); |
| 112 | + assert array.length == 16; |
| 113 | + for (byte b : array) { |
| 114 | + assert b == (byte) 0xFF; |
| 115 | + } |
| 116 | + } |
| 117 | + |
| 118 | + private static void retrieveMapColumn(Session session) { |
| 119 | + Row row = session.execute("SELECT b, m FROM examples.blobs WHERE k = 1").one(); |
| 120 | + |
| 121 | + // The map columns illustrates the pitfalls with position() and array(). |
| 122 | + Map<String, ByteBuffer> m = row.getMap("m", String.class, ByteBuffer.class); |
| 123 | + ByteBuffer buffer = m.get("test"); |
| 124 | + |
| 125 | + // We did get back a buffer that contains 16 bytes as expected. |
| 126 | + assert buffer.limit() - buffer.position() == 16; |
| 127 | + // However, it is not positioned at 0. And you can also see that its backing array contains more than 16 bytes. |
| 128 | + // What happens is that the buffer is a "view" of the last 16 of a 32-byte array. |
| 129 | + // This is an implementation detail and you shouldn't have to worry about it if you process the buffer correctly |
| 130 | + // (don't iterate from 0, use Bytes.getArray()). |
| 131 | + assert buffer.position() == 16; |
| 132 | + assert buffer.array().length == 32; |
| 133 | + } |
| 134 | + |
| 135 | + private static void insertConcurrent(Session session) { |
| 136 | + PreparedStatement preparedStatement = session.prepare("INSERT INTO examples.blobs (k, b) VALUES (1, :b)"); |
| 137 | + |
| 138 | + // This is another convenient utility provided by the driver. It's useful for tests. |
| 139 | + ByteBuffer buffer = Bytes.fromHexString("0xffffff"); |
| 140 | + |
| 141 | + // When you pass a byte buffer to a bound statement, it creates a shallow copy internally with the |
| 142 | + // buffer.duplicate() method. |
| 143 | + BoundStatement boundStatement = preparedStatement.bind(); |
| 144 | + boundStatement.setBytes("b", buffer); |
| 145 | + |
| 146 | + // This means you can now move in the original buffer, without affecting the insertion if it happens later. |
| 147 | + buffer.position(buffer.limit()); |
| 148 | + |
| 149 | + session.execute(boundStatement); |
| 150 | + Row row = session.execute("SELECT b FROM examples.blobs WHERE k = 1").one(); |
| 151 | + assert Bytes.toHexString(row.getBytes("b")).equals("0xffffff"); |
| 152 | + |
| 153 | + buffer.flip(); |
| 154 | + |
| 155 | + // HOWEVER duplicate() only performs a shallow copy. The two buffers still share the same contents. So if you |
| 156 | + // modify the contents of the original buffer, this will affect another execution of the bound statement. |
| 157 | + buffer.put(0, (byte) 0xaa); |
| 158 | + session.execute(boundStatement); |
| 159 | + row = session.execute("SELECT b FROM examples.blobs WHERE k = 1").one(); |
| 160 | + assert Bytes.toHexString(row.getBytes("b")).equals("0xaaffff"); |
| 161 | + |
| 162 | + // This will also happen if you use the async API, e.g. create the bound statement, call executeAsync() on it |
| 163 | + // and reuse the buffer immediately. |
| 164 | + |
| 165 | + // If you reuse buffers concurrently and want to avoid those issues, perform a deep copy of the buffer before |
| 166 | + // passing it to the bound statement. |
| 167 | + int startPosition = buffer.position(); |
| 168 | + ByteBuffer buffer2 = ByteBuffer.allocate(buffer.limit() - startPosition); |
| 169 | + buffer2.put(buffer); |
| 170 | + buffer.position(startPosition); |
| 171 | + buffer2.flip(); |
| 172 | + boundStatement.setBytes("b", buffer2); |
| 173 | + session.execute(boundStatement); |
| 174 | + |
| 175 | + // Note: unlike BoundStatement, SimpleStatement does not duplicate its arguments, so even the position will be |
| 176 | + // affected if you change it before executing the statement. Again, resort to deep copies if required. |
| 177 | + } |
| 178 | + |
| 179 | + private static void insertFromAndRetrieveToFile(Session session) throws IOException { |
| 180 | + ByteBuffer buffer = readAll(FILE); |
| 181 | + session.execute("INSERT INTO examples.blobs (k, b) VALUES (1, ?)", buffer); |
| 182 | + |
| 183 | + File tmpFile = File.createTempFile("blob", ".png"); |
| 184 | + System.out.printf("Writing retrieved buffer to %s%n", tmpFile.getAbsoluteFile()); |
| 185 | + |
| 186 | + Row row = session.execute("SELECT b FROM examples.blobs WHERE k = 1").one(); |
| 187 | + writeAll(row.getBytes("b"), tmpFile); |
| 188 | + } |
| 189 | + |
| 190 | + // Note: |
| 191 | + // - this is written with Java 6 APIs; if you're on a more recent version this can be improved (try-with-resources, |
| 192 | + // new-new io...) |
| 193 | + // - this reads the whole file in memory in one go. If your file does not fit in memory you should probably not |
| 194 | + // insert it into Cassandra either ;) |
| 195 | + private static ByteBuffer readAll(File file) throws IOException { |
| 196 | + FileInputStream inputStream = null; |
| 197 | + boolean threw = false; |
| 198 | + try { |
| 199 | + inputStream = new FileInputStream(file); |
| 200 | + FileChannel channel = inputStream.getChannel(); |
| 201 | + ByteBuffer buffer = ByteBuffer.allocate((int) channel.size()); |
| 202 | + channel.read(buffer); |
| 203 | + buffer.flip(); |
| 204 | + return buffer; |
| 205 | + } catch (IOException e) { |
| 206 | + threw = true; |
| 207 | + throw e; |
| 208 | + } finally { |
| 209 | + close(inputStream, threw); |
| 210 | + } |
| 211 | + } |
| 212 | + |
| 213 | + private static void writeAll(ByteBuffer buffer, File file) throws IOException { |
| 214 | + FileOutputStream outputStream = null; |
| 215 | + boolean threw = false; |
| 216 | + try { |
| 217 | + outputStream = new FileOutputStream(file); |
| 218 | + FileChannel channel = outputStream.getChannel(); |
| 219 | + channel.write(buffer); |
| 220 | + } catch (IOException e) { |
| 221 | + threw = true; |
| 222 | + throw e; |
| 223 | + } finally { |
| 224 | + close(outputStream, threw); |
| 225 | + } |
| 226 | + } |
| 227 | + |
| 228 | + private static void close(Closeable inputStream, boolean threw) throws IOException { |
| 229 | + if (inputStream != null) |
| 230 | + try { |
| 231 | + inputStream.close(); |
| 232 | + } catch (IOException e) { |
| 233 | + if (!threw) throw e; // else preserve original exception |
| 234 | + } |
| 235 | + } |
| 236 | +} |
0 commit comments