Refactor loaders to use inputstream allowing for potential use

for set binary stream when available to JDBC driver. Fixes #705 where Files.readAllBytes throws out of memory error. Conflicts: data-migration/src/test/java/com/quorum/tessera/data/migration/DirectoryStoreFileTest.java
Consensys · Apr 23, 2019 · bde2ef1 · bde2ef1
1 parent 4b9338e
commit bde2ef1
Show file tree

Hide file tree

Showing 17 changed files with 148 additions and 47 deletions.
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/BdbDumpFile.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/BdbDumpFile.java
@@ -1,7 +1,9 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.BufferedReader;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.*;
@@ -18,9 +20,9 @@
 public class BdbDumpFile implements StoreLoader {
 
     @Override
-    public Map<byte[], byte[]> load(Path inputFile) throws IOException {
+    public Map<byte[], InputStream> load(Path inputFile) throws IOException {
 
-        Map<byte[], byte[]> results = new HashMap<>();
+        Map<byte[], InputStream> results = new HashMap<>();
 
         try (BufferedReader reader = Files.newBufferedReader(inputFile)) {
 
@@ -38,8 +40,11 @@ public Map<byte[], byte[]> load(Path inputFile) throws IOException {
 
                 final String value = reader.readLine();
 
-
-                results.put(Base64.getDecoder().decode(Hex.decode(key)), Hex.decode(value));
+                InputStream inputStream = Optional.of(value)
+                    .map(Hex::decode)
+                    .map(ByteArrayInputStream::new)
+                    .get();
+                results.put(Base64.getDecoder().decode(Hex.decode(key)), inputStream);
             }
             return Collections.unmodifiableMap(results);
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/CmdLineExecutor.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/CmdLineExecutor.java
@@ -113,7 +113,7 @@ protected static int execute(String... args) throws Exception {
         final StoreLoader storeLoader = StoreLoader.create(storeType);
 
         final Path inputpath = Paths.get(line.getOptionValue("inputpath"));
-        final Map<byte[], byte[]> data = storeLoader.load(inputpath);
+        final Map<byte[], InputStream> data = storeLoader.load(inputpath);
 
         final String username = line.getOptionValue("dbuser");
         final String password = line.getOptionValue("dbpass");

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/DataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/DataExporter.java
@@ -1,11 +1,12 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.sql.SQLException;
 import java.util.Map;
 
 public interface DataExporter {
 
-    void export(Map<byte[], byte[]> data, Path output, String username, String password) throws SQLException;
+    void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException;
 
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/DirectoryStoreFile.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/DirectoryStoreFile.java
@@ -2,6 +2,7 @@
 
 import com.quorum.tessera.io.FilesDelegate;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Map;
@@ -15,7 +16,7 @@ public class DirectoryStoreFile implements StoreLoader {
     private final FilesDelegate fileDelegate = FilesDelegate.create();
 
     @Override
-    public Map<byte[], byte[]> load(Path directory) throws IOException {
+    public Map<byte[], InputStream> load(Path directory) throws IOException {
 
         Optional.ofNullable(directory)
                 .filter(p -> p.toFile().isDirectory())
@@ -24,7 +25,7 @@ public Map<byte[], byte[]> load(Path directory) throws IOException {
         try (Stream<Path> stream = Files.list(directory)) {
             return stream.collect(Collectors.toMap(
                     p -> new Base32().decode(p.toFile().getName()),
-                    p -> fileDelegate.readAllBytes(p)));
+                    p -> fileDelegate.newInputStream(p)));
         }
     }
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/H2DataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/H2DataExporter.java
@@ -1,5 +1,6 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
 import java.net.URL;
 import java.nio.file.Path;
 import java.sql.SQLException;
@@ -10,7 +11,7 @@ public class H2DataExporter implements DataExporter {
     private static final String INSERT_ROW = "INSERT INTO ENCRYPTED_TRANSACTION (HASH,ENCODED_PAYLOAD) VALUES (?,?)";
 
     @Override
-    public void export(final Map<byte[], byte[]> data,
+    public void export(final Map<byte[], InputStream> data,
                        final Path output,
                        final String username,
                        final String password) throws SQLException {

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/JdbcDataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/JdbcDataExporter.java
@@ -2,6 +2,7 @@
 
 import com.quorum.tessera.io.IOCallback;
 import com.quorum.tessera.io.UriCallback;
+import java.io.InputStream;
 
 import java.net.URL;
 import java.nio.file.Files;
@@ -29,7 +30,7 @@ public JdbcDataExporter(String jdbcUrl, String insertRow, URL ddl) {
     }
 
     @Override
-    public void export(Map<byte[], byte[]> data, Path output, String username, String password) throws SQLException {
+    public void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException {
 
         try (Connection conn = DriverManager.getConnection(jdbcUrl, username, password)) {
 
@@ -40,9 +41,9 @@ public void export(Map<byte[], byte[]> data, Path output, String username, Strin
             }
 
             try (PreparedStatement insertStatement = conn.prepareStatement(insertRow)) {
-                for (Entry<byte[], byte[]> values : data.entrySet()) {
+                for (Entry<byte[], InputStream> values : data.entrySet()) {
                     insertStatement.setBytes(1, values.getKey());
-                    insertStatement.setBytes(2, values.getValue());
+                    insertStatement.setBinaryStream(2, values.getValue());
                     insertStatement.execute();
                 }
             }

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteDataExporter.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteDataExporter.java
@@ -1,25 +1,53 @@
 package com.quorum.tessera.data.migration;
 
-import java.net.URL;
+import com.quorum.tessera.io.IOCallback;
+import com.quorum.tessera.io.UriCallback;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
 import java.util.Map;
 
 public class SqliteDataExporter implements DataExporter {
 
     private static final String INSERT_ROW = "INSERT INTO ENCRYPTED_TRANSACTION (HASH,ENCODED_PAYLOAD) VALUES (?,?)";
 
     @Override
-    public void export(Map<byte[], byte[]> data, Path output, final String username, final String password) throws SQLException {
+    public void export(Map<byte[], InputStream> data, Path output, String username, String password) throws SQLException {
 
         final String connectionString = "jdbc:sqlite:" + output.toString();
 
-        final URL sqlFile = getClass().getResource("/ddls/h2-ddl.sql");
+        final URI sqlFile = UriCallback.execute(() -> getClass().getResource("/ddls/sqlite-ddl.sql").toURI());
 
-        final JdbcDataExporter jdbcDataExporter = new JdbcDataExporter(connectionString, INSERT_ROW, sqlFile);
+        List<String> createTables = IOCallback.execute(() -> Files.readAllLines(Paths.get(sqlFile)));
 
-        jdbcDataExporter.export(data, output, username, password);
+        try (Connection conn = DriverManager.getConnection(connectionString, username, password)) {
 
+            try (Statement stmt = conn.createStatement()) {
+                for (String createTable : createTables) {
+                    stmt.executeUpdate(createTable);
+                }
+            }
+
+            try (PreparedStatement insertStatement = conn.prepareStatement(INSERT_ROW)) {
+                for (Map.Entry<byte[], InputStream> values : data.entrySet()) {
+                    insertStatement.setBytes(1, values.getKey());
+
+                    insertStatement.setBytes(2, Utils.toByteArray(values.getValue()));
+
+                    insertStatement.execute();
+                }
+            }
+
+        }
     }
 
+
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteLoader.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/SqliteLoader.java
@@ -1,6 +1,8 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -13,7 +15,7 @@
 public class SqliteLoader implements StoreLoader {
 
     @Override
-    public Map<byte[], byte[]> load(Path input) throws IOException {
+    public Map<byte[], InputStream> load(Path input) throws IOException {
 
         final String url = "jdbc:sqlite:" + input.toString();
 
@@ -23,11 +25,11 @@ public Map<byte[], byte[]> load(Path input) throws IOException {
                     Statement statement = conn.createStatement();
                     ResultSet results = statement.executeQuery("SELECT * FROM payload")) {
 
-                Map<byte[], byte[]> loadedData = new HashMap<>();
+                Map<byte[], InputStream> loadedData = new HashMap<>();
                 while (results.next()) {
 
                     byte[] key = results.getBytes("key");
-                    byte[] value = results.getBytes("bytes");
+                    InputStream value = new ByteArrayInputStream(results.getBytes("bytes"));
 
                     loadedData.put(key, value);
 

diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/StoreLoader.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/StoreLoader.java
@@ -1,6 +1,7 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Path;
 import java.util.Collections;
 import java.util.HashMap;
@@ -9,18 +10,23 @@
 
 public interface StoreLoader {
 
-    Map<byte[], byte[]> load(Path input) throws IOException;
+
+
+    Map<byte[], InputStream> load(Path input) throws IOException;
 
     Map<StoreType, StoreLoader> LOOKUP = Collections.unmodifiableMap(
-        new HashMap<StoreType, StoreLoader>() {{
+        new HashMap<StoreType, StoreLoader>() {
+        {
             put(StoreType.BDB, new BdbDumpFile());
             put(StoreType.DIR, new DirectoryStoreFile());
             put(StoreType.SQLITE, new SqliteLoader());
-        }}
+        }
+    }
     );
 
     static StoreLoader create(StoreType storeType) {
         return Optional.ofNullable(LOOKUP.get(storeType)).get();
     }
 
+
 }
diff --git a/data-migration/src/main/java/com/quorum/tessera/data/migration/Utils.java b/data-migration/src/main/java/com/quorum/tessera/data/migration/Utils.java
@@ -0,0 +1,22 @@
+package com.quorum.tessera.data.migration;
+
+import com.quorum.tessera.io.IOCallback;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+
+public interface Utils {
+
+    static byte[] toByteArray(InputStream in) {
+        return IOCallback.execute(() -> {
+            ByteArrayOutputStream os = new ByteArrayOutputStream();
+
+            byte[] buffer = new byte[1024];
+            int len;
+            while ((len = in.read(buffer)) != -1) {
+                os.write(buffer, 0, len);
+            }
+
+            return os.toByteArray();
+        });
+    }
+}
diff --git a/data-migration/src/test/java/com/quorum/tessera/data/migration/BdbDumpFileTest.java b/data-migration/src/test/java/com/quorum/tessera/data/migration/BdbDumpFileTest.java
@@ -1,6 +1,7 @@
 package com.quorum.tessera.data.migration;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -16,7 +17,7 @@ public void loadSample() throws URISyntaxException, IOException {
 
         Path inputFile = Paths.get(getClass().getResource("/bdb/bdb-sample.txt").toURI());
 
-        Map<byte[], byte[]> results = new BdbDumpFile().load(inputFile);
+        Map<byte[], InputStream> results = new BdbDumpFile().load(inputFile);
 
         assertThat(results).hasSize(12);
 
@@ -29,7 +30,7 @@ public void loadSimpleEntrySample() throws URISyntaxException, IOException {
 
         Path inputFile = Paths.get(getClass().getResource("/bdb/single-entry.txt").toURI());
 
-        Map<byte[], byte[]> results = new BdbDumpFile().load(inputFile);
+        Map<byte[], InputStream> results = new BdbDumpFile().load(inputFile);
 
         assertThat(results).hasSize(1);
 

diff --git a/data-migration/src/test/java/com/quorum/tessera/data/migration/DirectoryStoreFileTest.java b/data-migration/src/test/java/com/quorum/tessera/data/migration/DirectoryStoreFileTest.java
@@ -1,5 +1,7 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.InputStream;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Map;
@@ -14,11 +16,32 @@ public void load() throws Exception {
         Path directory = Paths.get(getClass().getResource("/dir/").toURI());
 
         DirectoryStoreFile directoryStoreFile = new DirectoryStoreFile();
-        
-        Map<byte[],byte[]> results = directoryStoreFile.load(directory);
-        
+
+        Map<byte[], InputStream> results = directoryStoreFile.load(directory);
+
         assertThat(results).hasSize(22);
 
     }
-
+
+    @Test
+    public void loadLarge() throws Exception {
+
+        Path baseDir = Paths.get(getClass().getResource("/").toURI());
+
+        Path directory = baseDir.resolve(UUID.randomUUID().toString());
+
+        Files.createDirectories(directory);
+
+        Path largeFile = Paths.get(directory.toAbsolutePath().toString(), "loadLarge");
+
+        Files.copy(getClass().getResourceAsStream("/loadLarge.sample"), largeFile);
+
+        DirectoryStoreFile directoryStoreFile = new DirectoryStoreFile();
+
+        Map<byte[], InputStream> results = directoryStoreFile.load(directory);
+
+        assertThat(results).hasSize(1);
+
+    }
+
 }
diff --git a/data-migration/src/test/java/com/quorum/tessera/data/migration/H2DataExporterTest.java b/data-migration/src/test/java/com/quorum/tessera/data/migration/H2DataExporterTest.java
@@ -1,5 +1,6 @@
 package com.quorum.tessera.data.migration;
 
+import java.io.ByteArrayInputStream;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -8,6 +9,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.sql.*;
@@ -52,8 +54,8 @@ public void exportSingleLine() throws SQLException, IOException {
 
         Path outputpath = Files.createTempFile("exportSingleLine", ".db");
 
-        Map<byte[], byte[]> singleLineData = new HashMap<>();
-        singleLineData.put("HASH".getBytes(), "VALUE".getBytes());
+        Map<byte[], InputStream> singleLineData = new HashMap<>();
+        singleLineData.put("HASH".getBytes(), new ByteArrayInputStream("VALUE".getBytes()));
 
         exporter.export(singleLineData, outputpath, null, null);
 
@@ -88,8 +90,8 @@ public void exportSingleLineWithUsernameAndPassword() throws SQLException, IOExc
 
         final Path outputpath = Files.createTempFile("exportSingleLine", ".db");
 
-        final Map<byte[], byte[]> singleLineData = new HashMap<>();
-        singleLineData.put("HASH".getBytes(), "VALUE".getBytes());
+        final Map<byte[], InputStream> singleLineData = new HashMap<>();
+        singleLineData.put("HASH".getBytes(), new ByteArrayInputStream("VALUE".getBytes()));
 
         exporter.export(singleLineData, outputpath, username, password);
 
@@ -124,8 +126,8 @@ public void exportSingleLineWithUsernameAndPasswordFailsWhenReading() throws SQL
 
         final Path outputpath = Files.createTempFile("exportSingleLine", ".db");
 
-        final Map<byte[], byte[]> singleLineData = new HashMap<>();
-        singleLineData.put("HASH".getBytes(), "VALUE".getBytes());
+        final Map<byte[], InputStream> singleLineData = new HashMap<>();
+        singleLineData.put("HASH".getBytes(), new ByteArrayInputStream("VALUE".getBytes()));
 
         exporter.export(singleLineData, outputpath, username, password);