Add flush mode support for JDBC sessions

Resolves: #1468

spring-session-jdbc/src/main/java/org/springframework/session/jdbc/JdbcOperationsSessionRepository.java

@@ -48,6 +48,7 @@ import org.springframework.jdbc.support.lob.DefaultLobHandler;
 import org.springframework.jdbc.support.lob.LobHandler;
 import org.springframework.session.DelegatingIndexResolver;
 import org.springframework.session.FindByIndexNameSessionRepository;
+import org.springframework.session.FlushMode;
 import org.springframework.session.IndexResolver;
 import org.springframework.session.MapSession;
 import org.springframework.session.PrincipalNameIndexResolver;
@@ -238,6 +239,8 @@ public class JdbcOperationsSessionRepository
 
 	private LobHandler lobHandler = new DefaultLobHandler();
 
+	private FlushMode flushMode = FlushMode.ON_SAVE;
+
 	private SaveMode saveMode = SaveMode.ON_SET_ATTRIBUTE;
 
 	/**
@@ -386,6 +389,15 @@ public class JdbcOperationsSessionRepository
 		this.conversionService = conversionService;
 	}
 
+	/**
+	 * Set the flush mode. Default is {@link FlushMode#ON_SAVE}.
+	 * @param flushMode the flush mode
+	 */
+	public void setFlushMode(FlushMode flushMode) {
+		Assert.notNull(flushMode, "flushMode must not be null");
+		this.flushMode = flushMode;
+	}
+
 	/**
 	 * Set the save mode.
 	 * @param saveMode the save mode
@@ -401,77 +413,14 @@ public class JdbcOperationsSessionRepository
 		if (this.defaultMaxInactiveInterval != null) {
 			delegate.setMaxInactiveInterval(Duration.ofSeconds(this.defaultMaxInactiveInterval));
 		}
-		return new JdbcSession(delegate, UUID.randomUUID().toString(), true);
+		JdbcSession session = new JdbcSession(delegate, UUID.randomUUID().toString(), true);
+		session.flushIfRequired();
+		return session;
 	}
 
 	@Override
 	public void save(final JdbcSession session) {
-		if (session.isNew()) {
+		session.save();
-			this.transactionOperations.execute(new TransactionCallbackWithoutResult() {
-
-				@Override
-				protected void doInTransactionWithoutResult(TransactionStatus status) {
-					Map<String, String> indexes = JdbcOperationsSessionRepository.this.indexResolver
-							.resolveIndexesFor(session);
-					JdbcOperationsSessionRepository.this.jdbcOperations
-							.update(JdbcOperationsSessionRepository.this.createSessionQuery, (ps) -> {
-								ps.setString(1, session.primaryKey);
-								ps.setString(2, session.getId());
-								ps.setLong(3, session.getCreationTime().toEpochMilli());
-								ps.setLong(4, session.getLastAccessedTime().toEpochMilli());
-								ps.setInt(5, (int) session.getMaxInactiveInterval().getSeconds());
-								ps.setLong(6, session.getExpiryTime().toEpochMilli());
-								ps.setString(7, indexes.get(PRINCIPAL_NAME_INDEX_NAME));
-							});
-					Set<String> attributeNames = session.getAttributeNames();
-					if (!attributeNames.isEmpty()) {
-						insertSessionAttributes(session, new ArrayList<>(attributeNames));
-					}
-				}
-
-			});
-		}
-		else {
-			this.transactionOperations.execute(new TransactionCallbackWithoutResult() {
-
-				@Override
-				protected void doInTransactionWithoutResult(TransactionStatus status) {
-					if (session.isChanged()) {
-						Map<String, String> indexes = JdbcOperationsSessionRepository.this.indexResolver
-								.resolveIndexesFor(session);
-						JdbcOperationsSessionRepository.this.jdbcOperations
-								.update(JdbcOperationsSessionRepository.this.updateSessionQuery, (ps) -> {
-									ps.setString(1, session.getId());
-									ps.setLong(2, session.getLastAccessedTime().toEpochMilli());
-									ps.setInt(3, (int) session.getMaxInactiveInterval().getSeconds());
-									ps.setLong(4, session.getExpiryTime().toEpochMilli());
-									ps.setString(5, indexes.get(PRINCIPAL_NAME_INDEX_NAME));
-									ps.setString(6, session.primaryKey);
-								});
-					}
-					List<String> addedAttributeNames = session.delta.entrySet().stream()
-							.filter((entry) -> entry.getValue() == DeltaValue.ADDED).map(Map.Entry::getKey)
-							.collect(Collectors.toList());
-					if (!addedAttributeNames.isEmpty()) {
-						insertSessionAttributes(session, addedAttributeNames);
-					}
-					List<String> updatedAttributeNames = session.delta.entrySet().stream()
-							.filter((entry) -> entry.getValue() == DeltaValue.UPDATED).map(Map.Entry::getKey)
-							.collect(Collectors.toList());
-					if (!updatedAttributeNames.isEmpty()) {
-						updateSessionAttributes(session, updatedAttributeNames);
-					}
-					List<String> removedAttributeNames = session.delta.entrySet().stream()
-							.filter((entry) -> entry.getValue() == DeltaValue.REMOVED).map(Map.Entry::getKey)
-							.collect(Collectors.toList());
-					if (!removedAttributeNames.isEmpty()) {
-						deleteSessionAttributes(session, removedAttributeNames);
-					}
-				}
-
-			});
-		}
-		session.clearChangeFlags();
 	}
 
 	@Override
@@ -806,6 +755,7 @@ public class JdbcOperationsSessionRepository
 			if (PRINCIPAL_NAME_INDEX_NAME.equals(attributeName) || SPRING_SECURITY_CONTEXT.equals(attributeName)) {
 				this.changed = true;
 			}
+			flushIfRequired();
 		}
 
 		@Override
@@ -822,6 +772,7 @@ public class JdbcOperationsSessionRepository
 		public void setLastAccessedTime(Instant lastAccessedTime) {
 			this.delegate.setLastAccessedTime(lastAccessedTime);
 			this.changed = true;
+			flushIfRequired();
 		}
 
 		@Override
@@ -833,6 +784,7 @@ public class JdbcOperationsSessionRepository
 		public void setMaxInactiveInterval(Duration interval) {
 			this.delegate.setMaxInactiveInterval(interval);
 			this.changed = true;
+			flushIfRequired();
 		}
 
 		@Override
@@ -845,6 +797,83 @@ public class JdbcOperationsSessionRepository
 			return this.delegate.isExpired();
 		}
 
+		private void flushIfRequired() {
+			if (JdbcOperationsSessionRepository.this.flushMode == FlushMode.IMMEDIATE) {
+				save();
+			}
+		}
+
+		private void save() {
+			if (this.isNew) {
+				JdbcOperationsSessionRepository.this.transactionOperations
+						.execute(new TransactionCallbackWithoutResult() {
+
+							@Override
+							protected void doInTransactionWithoutResult(TransactionStatus status) {
+								Map<String, String> indexes = JdbcOperationsSessionRepository.this.indexResolver
+										.resolveIndexesFor(JdbcSession.this);
+								JdbcOperationsSessionRepository.this.jdbcOperations
+										.update(JdbcOperationsSessionRepository.this.createSessionQuery, (ps) -> {
+											ps.setString(1, JdbcSession.this.primaryKey);
+											ps.setString(2, getId());
+											ps.setLong(3, getCreationTime().toEpochMilli());
+											ps.setLong(4, getLastAccessedTime().toEpochMilli());
+											ps.setInt(5, (int) getMaxInactiveInterval().getSeconds());
+											ps.setLong(6, getExpiryTime().toEpochMilli());
+											ps.setString(7, indexes.get(PRINCIPAL_NAME_INDEX_NAME));
+										});
+								Set<String> attributeNames = getAttributeNames();
+								if (!attributeNames.isEmpty()) {
+									insertSessionAttributes(JdbcSession.this, new ArrayList<>(attributeNames));
+								}
+							}
+
+						});
+			}
+			else {
+				JdbcOperationsSessionRepository.this.transactionOperations
+						.execute(new TransactionCallbackWithoutResult() {
+
+							@Override
+							protected void doInTransactionWithoutResult(TransactionStatus status) {
+								if (JdbcSession.this.changed) {
+									Map<String, String> indexes = JdbcOperationsSessionRepository.this.indexResolver
+											.resolveIndexesFor(JdbcSession.this);
+									JdbcOperationsSessionRepository.this.jdbcOperations
+											.update(JdbcOperationsSessionRepository.this.updateSessionQuery, (ps) -> {
+												ps.setString(1, getId());
+												ps.setLong(2, getLastAccessedTime().toEpochMilli());
+												ps.setInt(3, (int) getMaxInactiveInterval().getSeconds());
+												ps.setLong(4, getExpiryTime().toEpochMilli());
+												ps.setString(5, indexes.get(PRINCIPAL_NAME_INDEX_NAME));
+												ps.setString(6, JdbcSession.this.primaryKey);
+											});
+								}
+								List<String> addedAttributeNames = JdbcSession.this.delta.entrySet().stream()
+										.filter((entry) -> entry.getValue() == DeltaValue.ADDED).map(Map.Entry::getKey)
+										.collect(Collectors.toList());
+								if (!addedAttributeNames.isEmpty()) {
+									insertSessionAttributes(JdbcSession.this, addedAttributeNames);
+								}
+								List<String> updatedAttributeNames = JdbcSession.this.delta.entrySet().stream()
+										.filter((entry) -> entry.getValue() == DeltaValue.UPDATED)
+										.map(Map.Entry::getKey).collect(Collectors.toList());
+								if (!updatedAttributeNames.isEmpty()) {
+									updateSessionAttributes(JdbcSession.this, updatedAttributeNames);
+								}
+								List<String> removedAttributeNames = JdbcSession.this.delta.entrySet().stream()
+										.filter((entry) -> entry.getValue() == DeltaValue.REMOVED)
+										.map(Map.Entry::getKey).collect(Collectors.toList());
+								if (!removedAttributeNames.isEmpty()) {
+									deleteSessionAttributes(JdbcSession.this, removedAttributeNames);
+								}
+							}
+
+						});
+			}
+			clearChangeFlags();
+		}
+
 	}
 
 	private class SessionResultSetExtractor implements ResultSetExtractor<List<JdbcSession>> {

spring-session-jdbc/src/main/java/org/springframework/session/jdbc/config/annotation/web/http/EnableJdbcHttpSession.java

@@ -26,8 +26,11 @@ import javax.sql.DataSource;
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.session.FlushMode;
 import org.springframework.session.MapSession;
 import org.springframework.session.SaveMode;
+import org.springframework.session.Session;
+import org.springframework.session.SessionRepository;
 import org.springframework.session.config.annotation.web.http.EnableSpringHttpSession;
 import org.springframework.session.jdbc.JdbcOperationsSessionRepository;
 import org.springframework.session.web.http.SessionRepositoryFilter;
@@ -97,6 +100,18 @@ public @interface EnableJdbcHttpSession {
 	 */
 	String cleanupCron() default JdbcHttpSessionConfiguration.DEFAULT_CLEANUP_CRON;
 
+	/**
+	 * Flush mode for the sessions. The default is {@code ON_SAVE} which only updates the
+	 * backing database when {@link SessionRepository#save(Session)} is invoked. In a web
+	 * environment this happens just before the HTTP response is committed.
+	 * <p>
+	 * Setting the value to {@code IMMEDIATE} will ensure that the any updates to the
+	 * Session are immediately written to the database.
+	 * @return the flush mode
+	 * @since 2.2.0
+	 */
+	FlushMode flushMode() default FlushMode.ON_SAVE;
+
 	/**
 	 * Save mode for the session. The default is {@link SaveMode#ON_SET_ATTRIBUTE}, which
 	 * only saves changes made to session.

spring-session-jdbc/src/main/java/org/springframework/session/jdbc/config/annotation/web/http/JdbcHttpSessionConfiguration.java

@@ -42,6 +42,7 @@ import org.springframework.jdbc.support.lob.LobHandler;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.SchedulingConfigurer;
 import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+import org.springframework.session.FlushMode;
 import org.springframework.session.MapSession;
 import org.springframework.session.SaveMode;
 import org.springframework.session.config.annotation.web.http.SpringHttpSessionConfiguration;
@@ -78,6 +79,8 @@ public class JdbcHttpSessionConfiguration extends SpringHttpSessionConfiguration
 
 	private String cleanupCron = DEFAULT_CLEANUP_CRON;
 
+	private FlushMode flushMode = FlushMode.ON_SAVE;
+
 	private SaveMode saveMode = SaveMode.ON_SET_ATTRIBUTE;
 
 	private DataSource dataSource;
@@ -103,6 +106,7 @@ public class JdbcHttpSessionConfiguration extends SpringHttpSessionConfiguration
 			sessionRepository.setTableName(this.tableName);
 		}
 		sessionRepository.setDefaultMaxInactiveInterval(this.maxInactiveIntervalInSeconds);
+		sessionRepository.setFlushMode(this.flushMode);
 		sessionRepository.setSaveMode(this.saveMode);
 		if (this.lobHandler != null) {
 			sessionRepository.setLobHandler(this.lobHandler);
@@ -146,6 +150,10 @@ public class JdbcHttpSessionConfiguration extends SpringHttpSessionConfiguration
 		this.cleanupCron = cleanupCron;
 	}
 
+	public void setFlushMode(FlushMode flushMode) {
+		this.flushMode = flushMode;
+	}
+
 	public void setSaveMode(SaveMode saveMode) {
 		this.saveMode = saveMode;
 	}
@@ -207,6 +215,7 @@ public class JdbcHttpSessionConfiguration extends SpringHttpSessionConfiguration
 		if (StringUtils.hasText(cleanupCron)) {
 			this.cleanupCron = cleanupCron;
 		}
+		this.flushMode = attributes.getEnum("flushMode");
 		this.saveMode = attributes.getEnum("saveMode");
 	}
 

spring-session-jdbc/src/test/java/org/springframework/session/jdbc/JdbcOperationsSessionRepositoryTests.java

@@ -37,9 +37,11 @@ import org.springframework.security.authentication.UsernamePasswordAuthenticatio
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.authority.AuthorityUtils;
 import org.springframework.session.FindByIndexNameSessionRepository;
+import org.springframework.session.FlushMode;
 import org.springframework.session.MapSession;
 import org.springframework.session.SaveMode;
 import org.springframework.session.Session;
+import org.springframework.session.jdbc.JdbcOperationsSessionRepository.JdbcSession;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionDefinition;
 
@@ -57,6 +59,7 @@ import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
 
 /**
@@ -227,6 +230,12 @@ class JdbcOperationsSessionRepositoryTests {
 				.withMessage("conversionService must not be null");
 	}
 
+	@Test
+	void setFlushModeNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> this.repository.setFlushMode(null))
+				.withMessage("flushMode must not be null");
+	}
+
 	@Test
 	void setSaveModeNull() {
 		assertThatIllegalArgumentException().isThrownBy(() -> this.repository.setSaveMode(null))
@@ -254,6 +263,16 @@ class JdbcOperationsSessionRepositoryTests {
 		verifyZeroInteractions(this.jdbcOperations);
 	}
 
+	@Test
+	void createSessionImmediateFlushMode() {
+		this.repository.setFlushMode(FlushMode.IMMEDIATE);
+		JdbcSession session = this.repository.createSession();
+		assertThat(session.isNew()).isFalse();
+		assertPropagationRequiresNew();
+		verify(this.jdbcOperations).update(startsWith("INSERT"), isA(PreparedStatementSetter.class));
+		verifyNoMoreInteractions(this.jdbcOperations);
+	}
+
 	@Test
 	void saveNewWithoutAttributes() {
 		JdbcOperationsSessionRepository.JdbcSession session = this.repository.createSession();
@@ -773,6 +792,51 @@ class JdbcOperationsSessionRepositoryTests {
 		verifyZeroInteractions(this.jdbcOperations);
 	}
 
+	@Test
+	void flushModeImmediateSetAttribute() {
+		this.repository.setFlushMode(FlushMode.IMMEDIATE);
+		JdbcSession session = this.repository.new JdbcSession(new MapSession(), "primaryKey", false);
+		String attrName = "someAttribute";
+		session.setAttribute(attrName, "someValue");
+		assertPropagationRequiresNew();
+		verify(this.jdbcOperations).update(startsWith("INSERT INTO SPRING_SESSION_ATTRIBUTES("),
+				isA(PreparedStatementSetter.class));
+		verifyNoMoreInteractions(this.jdbcOperations);
+	}
+
+	@Test
+	void flushModeImmediateRemoveAttribute() {
+		this.repository.setFlushMode(FlushMode.IMMEDIATE);
+		MapSession cached = new MapSession();
+		cached.setAttribute("attribute1", "value1");
+		JdbcSession session = this.repository.new JdbcSession(cached, "primaryKey", false);
+		session.removeAttribute("attribute1");
+		assertPropagationRequiresNew();
+		verify(this.jdbcOperations).update(startsWith("DELETE FROM SPRING_SESSION_ATTRIBUTES WHERE"),
+				isA(PreparedStatementSetter.class));
+		verifyNoMoreInteractions(this.jdbcOperations);
+	}
+
+	@Test
+	void flushModeSetMaxInactiveIntervalInSeconds() {
+		this.repository.setFlushMode(FlushMode.IMMEDIATE);
+		JdbcSession session = this.repository.new JdbcSession(new MapSession(), "primaryKey", false);
+		session.setMaxInactiveInterval(Duration.ofSeconds(1));
+		assertPropagationRequiresNew();
+		verify(this.jdbcOperations).update(startsWith("UPDATE SPRING_SESSION SET"), isA(PreparedStatementSetter.class));
+		verifyNoMoreInteractions(this.jdbcOperations);
+	}
+
+	@Test
+	void flushModeSetLastAccessedTime() {
+		this.repository.setFlushMode(FlushMode.IMMEDIATE);
+		JdbcSession session = this.repository.new JdbcSession(new MapSession(), "primaryKey", false);
+		session.setLastAccessedTime(Instant.now());
+		assertPropagationRequiresNew();
+		verify(this.jdbcOperations).update(startsWith("UPDATE SPRING_SESSION SET"), isA(PreparedStatementSetter.class));
+		verifyNoMoreInteractions(this.jdbcOperations);
+	}
+
 	private void assertPropagationRequiresNew() {
 		ArgumentCaptor<TransactionDefinition> argument = ArgumentCaptor.forClass(TransactionDefinition.class);
 		verify(this.transactionManager, atLeastOnce()).getTransaction(argument.capture());

spring-session-jdbc/src/test/java/org/springframework/session/jdbc/config/annotation/web/http/JdbcHttpSessionConfigurationTests.java

@@ -31,6 +31,7 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.support.lob.LobHandler;
 import org.springframework.mock.env.MockEnvironment;
+import org.springframework.session.FlushMode;
 import org.springframework.session.SaveMode;
 import org.springframework.session.jdbc.JdbcOperationsSessionRepository;
 import org.springframework.session.jdbc.config.annotation.SpringSessionDataSource;
@@ -136,6 +137,20 @@ class JdbcHttpSessionConfigurationTests {
 		assertThat(ReflectionTestUtils.getField(configuration, "cleanupCron")).isEqualTo(CLEANUP_CRON_EXPRESSION);
 	}
 
+	@Test
+	void customFlushModeAnnotation() {
+		registerAndRefresh(DataSourceConfiguration.class, CustomFlushModeExpressionAnnotationConfiguration.class);
+		assertThat(this.context.getBean(JdbcHttpSessionConfiguration.class)).hasFieldOrPropertyWithValue("flushMode",
+				FlushMode.IMMEDIATE);
+	}
+
+	@Test
+	void customFlushModeSetter() {
+		registerAndRefresh(DataSourceConfiguration.class, CustomFlushModeExpressionSetterConfiguration.class);
+		assertThat(this.context.getBean(JdbcHttpSessionConfiguration.class)).hasFieldOrPropertyWithValue("flushMode",
+				FlushMode.IMMEDIATE);
+	}
+
 	@Test
 	void customSaveModeAnnotation() {
 		registerAndRefresh(DataSourceConfiguration.class, CustomSaveModeExpressionAnnotationConfiguration.class);
@@ -315,6 +330,20 @@ class JdbcHttpSessionConfigurationTests {
 
 	}
 
+	@EnableJdbcHttpSession(flushMode = FlushMode.IMMEDIATE)
+	static class CustomFlushModeExpressionAnnotationConfiguration {
+
+	}
+
+	@Configuration
+	static class CustomFlushModeExpressionSetterConfiguration extends JdbcHttpSessionConfiguration {
+
+		CustomFlushModeExpressionSetterConfiguration() {
+			setFlushMode(FlushMode.IMMEDIATE);
+		}
+
+	}
+
 	@EnableJdbcHttpSession(saveMode = SaveMode.ALWAYS)
 	static class CustomSaveModeExpressionAnnotationConfiguration {