Refactor mysql, postgres and sqlserver blob resource implementations …

…to use the respective streaming APIs

spring-content-jpa/pom.xml

@@ -33,6 +33,13 @@
   			<artifactId>commons-logging</artifactId>
   			<version>1.2</version>
   		</dependency>
+		<dependency>
+			<groupId>com.microsoft.sqlserver</groupId>
+			<artifactId>mssql-jdbc</artifactId>
+			<version>6.1.0.jre8</version>
+			<optional>true</optional>
+		</dependency>
+
 
 		<!-- Test Dependencies -->
 		<dependency>

spring-content-jpa/src/main/asciidoc/jpa.adoc

@@ -10,7 +10,7 @@ Spring Content JPA Stores are enabled with the following Java Config.
 ----
 @Configuration
 @EnableJpaRepositories
-@EnableJpaContentRepositories
+@EnableJpaStores
 @EnableTransactionManagement
 public static class ApplicationConfig {  
 
@@ -46,27 +46,51 @@ public static class ApplicationConfig {
 ----
 ====
 
-This configuration class sets up an embedded HSQL database using the EmbeddedDatabaseBuilder API from spring-jdbc.  We then set up an EntityManagerFactory and use Hibernate as the persistence provider.  The last infrastructure component declared here is the JpaTransactionManager.  
+This configuration class sets up an embedded HSQL database using the EmbeddedDatabaseBuilder API from spring-jdbc.  We
+then set up an EntityManagerFactory and use Hibernate as the persistence provider.  The last infrastructure component
+declared here is the JpaTransactionManager.
 
-We activate Spring Data JPA repositories using the @EnableJpaRepositories annotation,  Spring Content JPA stores using @EnableJpaContentRepositories and enable transaction management with the @EnableTransactionManagement annotation.
+We activate Spring Data JPA repositories using the `@EnableJpaRepositories` annotation.  We activate Spring Content JPA
+stores using `@EnableJpaStores` and enable transaction management with the `@EnableTransactionManagement` annotation.
 
-If no base packages are configured both Spring Data JPA and Spring Content JPA will use the one the configuration class resides in.
+If no base packages are configured both Spring Data JPA and Spring Content JPA will use the package that the
+configuration class resides in as the base package.
+
+=== Content Streaming
+
+Spring Content JPA attempts to provide streams that chunk BLOBs through the memory space of the JVM rather that loading
+the entire BLOB into memory.  In order to do this the following database-specific implementation are used.
+
+==== MySQL
+
+The MySQL implementation requires MySQL's locator emulation to be turned on.  Your JDBC connection string must
+include the `emulateLocators=true` parameter.
+
+=== Postgresql
+
+The Postgresql implementation uses the OID field type and the Large Object API.  Whilst this provides properly chuncked
+streams it does comes with a significant performance disadvantage over bytea fields.
+
+=== SQL Server
+
+The SQL Server implementation uses [adaptive buffering](https://docs.microsoft.com/en-us/sql/connect/jdbc/using-adaptive-buffering?view=sql-server-2017#setting-adaptive-buffering)
+to serve BLOBs in a memory efficient way.
 
 == Persisting Content
 
 === Setting Content
 
 Storing content is achieved using the `ContentStore.setContent(entity, InputStream)` method.  
 
-The Spring Content Commons provided `PlacementService` is used to determine the actual physical storage location by parsing the @ContentId field into a resource path.
+The entity's `@ContentId` and `@ContentLength` fields will be updated.
 
-The `entity` @ContentId and @ContentLength fields will be updated.  
-
-If content has been previously stored this will overwritten with the new content updating just the @ContentLength field, if appropriate.
+If content has been previously stored this will overwritten with the new content updating just the `@ContentLength`
+field, if appropriate.
 
 ==== How the @ContentId field is handled 
 
-Spring Data JPA requires that content entities have an `@ContentId` field for identity that will be generated when content is initially set.  
+Spring Data JPA requires that content entities have an `@ContentId` field for identity that will be generated when
+content is initially set.
 
 === Getting Content
 
@@ -75,3 +99,4 @@ Content can be accessed using the `ContentStore.getContent(entity)` method.
 === Unsetting Content
 
 Content can be removed using the `ContentStore.unsetContent(entity)` method.
+

spring-content-jpa/src/main/java/internal/org/springframework/content/jpa/config/JpaStoreConfiguration.java

@@ -1,7 +1,10 @@
 package internal.org.springframework.content.jpa.config;
 
+import internal.org.springframework.content.jpa.io.MySQLBlobResource;
+import internal.org.springframework.content.jpa.io.SQLServerBlobResource;
+import org.springframework.content.jpa.io.CustomizableBlobResourceLoader;
 import internal.org.springframework.content.jpa.io.DelegatingBlobResourceLoader;
-import internal.org.springframework.content.jpa.io.GenericBlobResourceLoader;
+import internal.org.springframework.content.jpa.io.GenericBlobResource;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.content.jpa.io.BlobResourceLoader;
 import org.springframework.context.annotation.Bean;
@@ -25,8 +28,17 @@ public DelegatingBlobResourceLoader blobResourceLoader(DataSource ds,
 	}
 
 	@Bean
-	public BlobResourceLoader genericBlobResourceLoader(DataSource ds,
-			PlatformTransactionManager txnMgr) {
-		return new GenericBlobResourceLoader(new JdbcTemplate(ds), txnMgr);
+	public BlobResourceLoader genericBlobResourceLoader(DataSource ds, PlatformTransactionManager txnMgr) {
+		return new CustomizableBlobResourceLoader(new JdbcTemplate(ds), txnMgr, "GENERIC", (l, t, txn) -> { return new GenericBlobResource(l, t, txn); });
+	}
+
+	@Bean
+	public BlobResourceLoader mysqlBlobResourceLoader(DataSource ds, PlatformTransactionManager txnMgr) {
+		return new CustomizableBlobResourceLoader(new JdbcTemplate(ds), txnMgr, "MySQL", (l, t, txn) -> { return new MySQLBlobResource(l, t, txn); });
+	}
+
+	@Bean
+	public BlobResourceLoader sqlServerBlobResourceLoader(DataSource ds, PlatformTransactionManager txnMgr) {
+		return new CustomizableBlobResourceLoader(new JdbcTemplate(ds), txnMgr, "Microsoft SQL Server", (l, t, txn) -> { return new SQLServerBlobResource(l, t, txn); });
 	}
 }

spring-content-jpa/src/main/java/internal/org/springframework/content/jpa/io/MySQLBlobResource.java

@@ -0,0 +1,17 @@
+package internal.org.springframework.content.jpa.io;
+
+import org.springframework.content.jpa.io.AbstractBlobResource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+
+public class MySQLBlobResource extends AbstractBlobResource {
+
+	public MySQLBlobResource(Object id, JdbcTemplate template, PlatformTransactionManager txnMgr) {
+		super(id, template, txnMgr);
+	}
+
+	@Override
+	protected String getSelectBlobSQL(Object id) {
+		return "SELECT id, 'content' as content FROM BLOBS WHERE id='" + id + "'";
+	}
+}

spring-content-jpa/src/main/java/internal/org/springframework/content/jpa/io/ResourceProvider.java

@@ -0,0 +1,11 @@
+package internal.org.springframework.content.jpa.io;
+
+import org.springframework.core.io.Resource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+
+public interface ResourceProvider {
+
+    Resource getResource(Object id, JdbcTemplate template, PlatformTransactionManager txnMgr);
+
+}

spring-content-jpa/src/main/java/internal/org/springframework/content/jpa/io/SQLServerBlobResource.java

@@ -0,0 +1,63 @@
+package internal.org.springframework.content.jpa.io;
+
+import com.microsoft.sqlserver.jdbc.SQLServerStatement;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.content.jpa.io.AbstractBlobResource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import javax.sql.DataSource;
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static java.lang.String.format;
+
+public class SQLServerBlobResource extends AbstractBlobResource {
+
+	private static Log logger = LogFactory.getLog(SQLServerBlobResource.class);
+
+	public SQLServerBlobResource(Object id, JdbcTemplate template, PlatformTransactionManager txnMgr) {
+		super(id, template, txnMgr);
+	}
+
+	@Override
+	public InputStream getInputStream() throws IOException {
+		final Object id = getId();
+
+		String sql = getSelectBlobSQL(getId());
+
+		DataSource ds = getTemplate().getDataSource();
+		Connection conn = DataSourceUtils.getConnection(ds);
+		try {
+			conn.setAutoCommit(false);
+		} catch (SQLException e) {
+			logger.error(format("setting autocommit to false whilst getting content %s", id), e);
+		}
+		InputStream is = null;
+		Statement stmt = null;
+		ResultSet rs = null;
+		try {
+			stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_UPDATABLE);
+
+			if (stmt.isWrapperFor(com.microsoft.sqlserver.jdbc.SQLServerStatement.class)) {
+				SQLServerStatement SQLstmt = stmt.unwrap(com.microsoft.sqlserver.jdbc.SQLServerStatement.class);
+				SQLstmt.setResponseBuffering("adaptive");
+			}
+			rs = stmt.executeQuery(sql);
+
+			if (!rs.next())
+				return null;
+			is = rs.getBinaryStream(2);
+		}
+		catch (SQLException e) {
+			logger.error(format("getting content %s", id), e);
+		}
+		return new ClosingInputStream(id, is, rs, stmt, conn, ds);
+	}
+}

spring-content-jpa/src/main/java/org/springframework/content/jpa/io/AbstractBlobResource.java

@@ -51,6 +51,10 @@ protected void setId(Object id) {
 		}
 	}
 
+	protected JdbcTemplate getTemplate() {
+		return template;
+	}
+
 	@Override
 	public boolean isWritable() {
 		return true;
@@ -126,7 +130,7 @@ public String doInPreparedStatement(PreparedStatement ps)
 							// mark the resource as being updated
 							resource.setId(-1);
 
-							ps.setBinaryStream(1, fin);
+							ps.setBlob(1, fin);
 							ps.setString(2, id.toString());
 							ps.executeUpdate();
 							IOUtils.closeQuietly(fin);
@@ -150,8 +154,9 @@ public String doInPreparedStatement(PreparedStatement ps)
 							ResultSet set = null;
 							try {
 								ps.setString(1, rid.toString());
-								ps.setBinaryStream(2, fin);
+								ps.setBlob(2, fin);
 								ps.executeUpdate();
+								IOUtils.closeQuietly(fin);
 								return rid.toString();
 							}
 							catch (SQLException sqle) {
@@ -248,10 +253,16 @@ public String getDescription() {
 	@Override
 	public InputStream getInputStream() throws IOException {
 		final Object id = this.id;
-		String sql = "SELECT content FROM BLOBS WHERE id='" + this.id + "'";
+
+		String sql = getSelectBlobSQL(this.id);
 
 		DataSource ds = this.template.getDataSource();
 		Connection conn = DataSourceUtils.getConnection(ds);
+		try {
+			conn.setAutoCommit(false);
+		} catch (SQLException e) {
+			logger.error(format("getting content %s", id), e);
+		}
 		InputStream is = null;
 		Statement stmt = null;
 		ResultSet rs = null;
@@ -260,7 +271,8 @@ public InputStream getInputStream() throws IOException {
 			rs = stmt.executeQuery(sql);
 			if (!rs.next())
 				return null;
-			is = rs.getBinaryStream(1);
+			Blob b = rs.getBlob(2);
+			is = b.getBinaryStream();
 		}
 		catch (SQLException e) {
 			logger.error(format("getting content %s", id), e);
@@ -275,6 +287,11 @@ public void delete() throws IOException {
 		this.template.update(sql);
 	}
 
+
+	protected String getSelectBlobSQL(Object id) {
+		return "SELECT id, content FROM BLOBS WHERE id='" + id + "'";
+	}
+
 	public class ClosingInputStream extends InputStream {
 
 		private Object id;

spring-content-jpa/src/main/java/org/springframework/content/jpa/io/CustomizableBlobResourceLoader.java

@@ -0,0 +1,48 @@
+package org.springframework.content.jpa.io;
+
+import internal.org.springframework.content.jpa.io.GenericBlobResource;
+import internal.org.springframework.content.jpa.io.ResourceProvider;
+import org.springframework.content.jpa.io.BlobResourceLoader;
+import org.springframework.core.io.Resource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.ClassUtils;
+
+import java.util.function.Function;
+
+public class CustomizableBlobResourceLoader implements BlobResourceLoader {
+
+	private JdbcTemplate template;
+	private PlatformTransactionManager txnMgr;
+	private String databaseName;
+	private ResourceProvider resourceProvider;
+
+	public CustomizableBlobResourceLoader(JdbcTemplate template, PlatformTransactionManager txnMgr) {
+		this.template = template;
+		this.txnMgr = txnMgr;
+		this.databaseName = "GENERIC";
+		this.resourceProvider = (l, t, txn) -> { return new GenericBlobResource(l, t, txn);};
+	}
+
+	public CustomizableBlobResourceLoader(JdbcTemplate template, PlatformTransactionManager txnMgr, String databaseName, ResourceProvider resourceProvider) {
+		this.template = template;
+		this.txnMgr = txnMgr;
+		this.databaseName = databaseName;
+		this.resourceProvider = resourceProvider;
+	}
+
+	@Override
+	public String getDatabaseName() {
+		return databaseName;
+	}
+
+	@Override
+	public Resource getResource(String location) {
+		return resourceProvider.getResource(location, template, txnMgr);
+	}
+
+	@Override
+	public ClassLoader getClassLoader() {
+		return ClassUtils.getDefaultClassLoader();
+	}
+}

spring-content-jpa/src/main/resources/org/springframework/content/jpa/schema-postgresql.sql

@@ -1,5 +1,5 @@
 CREATE TABLE IF NOT EXISTS BLOBS (
 	id SERIAL PRIMARY KEY,
-	content bytea
+	content oid
 );
-ALTER TABLE BLOBS ALTER COLUMN id TYPE VARCHAR(36)
+ALTER TABLE BLOBS ALTER COLUMN id TYPE VARCHAR(36)

spring-content-jpa/src/main/resources/org/springframework/content/jpa/schema-sqlserver.sql

@@ -1,3 +1,2 @@
-IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='BLOBS' AND xtype='U') CREATE TABLE BLOBS ( id VARCHAR(36) NOT NULL, content varBinary(MAX) )
-ALTER TABLE BLOBS ADD CONSTRAINT pk_id PRIMARY KEY (id)
-
+IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='BLOBS' AND xtype='U') CREATE TABLE BLOBS ( id VARCHAR(36) NOT NULL, content varBinary(MAX) );
+ALTER TABLE BLOBS ADD CONSTRAINT pk_id PRIMARY KEY (id);