Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make LockRegistry#obtain Java 8 based

Browse Source 
This commit harmonizes `LockRegistry#obtain` logic in Zookeeper and JDBC
implementations with Redis implementation
by using Java 8 `ConcurrentMap#computeIfAbsent`

* Remove `synchronized (this.locks)` from the `expireUnusedOlderThan`
implementations because `iterator()` is thread-safe from `ConcurrentHashMap`
* Fix deprecation in the `IntegrationGraphControllerRegistrar`
* Revert Spring Security version to `4.2.2`, since `5.0 B-S` is broken
This commit is contained in:
Vedran Pavic
2017-04-21 21:16:41 +02:00
committed by Artem Bilan
parent 7f2c1d6ecd
commit 1dc3726e7d
6 changed files with 59 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -3,7 +3,7 @@ buildscript {
maven { url 'https://repo.spring.io/plugins-release' }
}
dependencies {
classpath 'io.spring.gradle:dependency-management-plugin:1.0.0.RC2'
classpath 'io.spring.gradle:dependency-management-plugin:1.0.2.RELEASE'
classpath 'io.spring.gradle:spring-io-plugin:0.0.6.RELEASE'
classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.0'
@@ -135,7 +135,7 @@ subprojects { subproject ->
springDataMongoVersion = '2.0.0.BUILD-SNAPSHOT'
springDataRedisVersion = '2.0.0.BUILD-SNAPSHOT'
springGemfireVersion = '2.0.0.BUILD-SNAPSHOT'
springSecurityVersion = '5.0.0.BUILD-SNAPSHOT'
springSecurityVersion = '4.2.2.RELEASE'
springSocialTwitterVersion = '2.0.0.M1'
springRetryVersion = '1.2.0.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.0.BUILD-SNAPSHOT'

7
spring-integration-http/src/main/java/org/springframework/integration/http/config/IntegrationGraphControllerRegistrar.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -40,13 +40,14 @@ import org.springframework.integration.http.management.IntegrationGraphControlle
import org.springframework.integration.http.support.HttpContextUtils;
import org.springframework.integration.support.management.graph.IntegrationGraphServer;
import org.springframework.web.servlet.config.annotation.CorsRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
/**
* Registers the necessary beans for {@link EnableIntegrationGraphController}.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 4.3
*/
class IntegrationGraphControllerRegistrar implements ImportBeanDefinitionRegistrar {
@@ -113,7 +114,7 @@ class IntegrationGraphControllerRegistrar implements ImportBeanDefinitionRegistr
}
private static final class IntegrationGraphCorsConfigurer extends WebMvcConfigurerAdapter {
private static final class IntegrationGraphCorsConfigurer implements WebMvcConfigurer {
private final String path;

34
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,10 +16,10 @@
package org.springframework.integration.jdbc.lock;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@@ -50,7 +50,7 @@ import org.springframework.util.Assert;
*/
public class JdbcLockRegistry implements ExpirableLockRegistry {
private final Map<String, JdbcLock> locks = new HashMap<String, JdbcLock>();
private final Map<String, JdbcLock> locks = new ConcurrentHashMap<>();
private final LockRepository client;
@@ -62,17 +62,7 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
public Lock obtain(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = pathFor((String) lockKey);
JdbcLock lock = this.locks.get(path);
if (lock == null) {
synchronized (this.locks) {
lock = this.locks.get(path);
if (lock == null) {
lock = new JdbcLock(this.client, path);
this.locks.put(path, lock);
}
}
}
return lock;
return this.locks.computeIfAbsent(path, p -> new JdbcLock(this.client, p));
}
private String pathFor(String input) {
@@ -81,15 +71,13 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
@Override
public void expireUnusedOlderThan(long age) {
synchronized (this.locks) {
Iterator<Entry<String, JdbcLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, JdbcLock> entry = iterator.next();
JdbcLock lock = entry.getValue();
if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
Iterator<Entry<String, JdbcLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, JdbcLock> entry = iterator.next();
JdbcLock lock = entry.getValue();
if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
}
}

17
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsInboundGatewayParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -63,7 +63,7 @@ public class JmsInboundGatewayParserTests {
JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("jmsGateway");
assertEquals(JmsMessageDrivenEndpoint.class, gateway.getClass());
context.start();
Message<?> message = channel.receive(3000);
Message<?> message = channel.receive(10000);
MessageHistory history = MessageHistory.read(message);
assertNotNull(history);
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "jmsGateway", 0);
@@ -82,7 +82,7 @@ public class JmsInboundGatewayParserTests {
JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("jmsGateway");
assertEquals(JmsMessageDrivenEndpoint.class, gateway.getClass());
context.start();
Message<?> message = channel.receive(3000);
Message<?> message = channel.receive(10000);
assertNotNull("message should not be null", message);
assertEquals("message-driven-test", message.getPayload());
context.close();
@@ -96,7 +96,7 @@ public class JmsInboundGatewayParserTests {
JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("jmsGateway");
assertEquals(JmsMessageDrivenEndpoint.class, gateway.getClass());
context.start();
Message<?> message = channel.receive(3000);
Message<?> message = channel.receive(10000);
assertNotNull("message should not be null", message);
assertEquals("converted-test-message", message.getPayload());
context.close();
@@ -188,7 +188,7 @@ public class JmsInboundGatewayParserTests {
JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("jmsGateway");
assertEquals(JmsMessageDrivenEndpoint.class, gateway.getClass());
context.start();
Message<?> message = channel.receive(3000);
Message<?> message = channel.receive(10000);
assertNotNull("message should not be null", message);
assertEquals("message-driven-test", message.getPayload());
context.close();
@@ -387,7 +387,8 @@ public class JmsInboundGatewayParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsGatewayWithPubSubDomain.xml", this.getClass());
JmsMessageDrivenEndpoint endpoint = context.getBean("gateway", JmsMessageDrivenEndpoint.class);
JmsDestinationAccessor container = (JmsDestinationAccessor) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer");
JmsDestinationAccessor container =
(JmsDestinationAccessor) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer");
assertEquals(Boolean.TRUE, container.isPubSubDomain());
context.close();
}
@@ -397,7 +398,9 @@ public class JmsInboundGatewayParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"inboundGatewayWithDurableSubscription.xml", this.getClass());
JmsMessageDrivenEndpoint endpoint = context.getBean("gateway", JmsMessageDrivenEndpoint.class);
DefaultMessageListenerContainer container = (DefaultMessageListenerContainer) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer");
DefaultMessageListenerContainer container =
(DefaultMessageListenerContainer) new DirectFieldAccessor(endpoint)
.getPropertyValue("listenerContainer");
assertEquals(Boolean.TRUE, container.isPubSubDomain());
assertEquals(Boolean.TRUE, container.isSubscriptionDurable());
assertEquals("testDurableSubscriptionName", container.getDurableSubscriptionName());

35
spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
View File

@@ -23,7 +23,6 @@ import java.util.Iterator;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
@@ -77,16 +76,16 @@ public final class RedisLockRegistry implements ExpirableLockRegistry {
private static final String OBTAIN_LOCK_SCRIPT =
"local lockClientId = redis.call('GET', KEYS[1])\n" +
"if lockClientId == ARGV[1] then\n" +
" redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" +
" return true\n" +
"elseif not lockClientId then\n" +
" redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])\n" +
" return true\n" +
"end\n" +
"return false";
"if lockClientId == ARGV[1] then\n" +
" redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" +
" return true\n" +
"elseif not lockClientId then\n" +
" redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])\n" +
" return true\n" +
"end\n" +
"return false";
private final ConcurrentMap<String, RedisLock> locks = new ConcurrentHashMap<>();
private final Map<String, RedisLock> locks = new ConcurrentHashMap<>();
private final String clientId = UUID.randomUUID().toString();
@@ -131,15 +130,13 @@ public final class RedisLockRegistry implements ExpirableLockRegistry {
@Override
public void expireUnusedOlderThan(long age) {
synchronized (this.locks) {
Iterator<Map.Entry<String, RedisLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Map.Entry<String, RedisLock> entry = iterator.next();
RedisLock lock = entry.getValue();
if (now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
Iterator<Map.Entry<String, RedisLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Map.Entry<String, RedisLock> entry = iterator.next();
RedisLock lock = entry.getValue();
if (now - lock.getLockedAt() > age && !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
}
}

41
spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@
package org.springframework.integration.zookeeper.lock;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -45,6 +45,8 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author Artem Bilan
* @author Vedran Pavic
*
* @since 4.2
*
*/
@@ -56,7 +58,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
private final KeyToPathStrategy keyToPath;
private final Map<String, ZkLock> locks = new HashMap<String, ZkLock>();
private final Map<String, ZkLock> locks = new ConcurrentHashMap<>();
private final boolean trackingTime;
@@ -124,18 +126,9 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
public Lock obtain(Object lockKey) {
Assert.isInstanceOf(String.class, lockKey);
String path = this.keyToPath.pathFor((String) lockKey);
ZkLock lock = this.locks.get(path);
if (lock == null) {
synchronized (this.locks) {
lock = this.locks.get(path);
if (lock == null) {
lock = new ZkLock(this.client, this.mutexTaskExecutor, path);
this.locks.put(path, lock);
}
if (this.trackingTime) {
lock.setLastUsed(System.currentTimeMillis());
}
}
ZkLock lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, this.mutexTaskExecutor, p));
if (this.trackingTime) {
lock.setLastUsed(System.currentTimeMillis());
}
return lock;
}
@@ -152,16 +145,14 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
if (!this.trackingTime) {
throw new IllegalStateException("Ths KeyToPathStrategy is bounded; expiry is not supported");
}
synchronized (this.locks) {
Iterator<Entry<String, ZkLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, ZkLock> entry = iterator.next();
ZkLock lock = entry.getValue();
if (now - lock.getLastUsed() > age
&& !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
Iterator<Entry<String, ZkLock>> iterator = this.locks.entrySet().iterator();
long now = System.currentTimeMillis();
while (iterator.hasNext()) {
Entry<String, ZkLock> entry = iterator.next();
ZkLock lock = entry.getValue();
if (now - lock.getLastUsed() > age
&& !lock.isAcquiredInThisProcess()) {
iterator.remove();
}
}
}