Distributed lock

settings.gradle

@@ -37,6 +37,10 @@ include 'spring-statemachine-data:jpa'
 include 'spring-statemachine-data:redis'
 include 'spring-statemachine-data:mongodb'
 
+include 'spring-statemachine-lock'
+include 'spring-statemachine-lock:shedlock'
+include 'spring-statemachine-lock:redisson'
+
 rootProject.children.find {
 	if (it.name == 'spring-statemachine-recipes') {
 		it.name = 'spring-statemachine-recipes-common'
@@ -53,5 +57,11 @@ rootProject.children.find {
 			it.name = 'spring-statemachine-data-' + it.name
 		}
 	}
+	if (it.name == 'spring-statemachine-lock') {
+		it.name = 'spring-statemachine-lock-common'
+		it.children.each {
+			it.name = 'spring-statemachine-lock-' + it.name
+		}
+	}
 }
 

spring-statemachine-lock/build.gradle

@@ -0,0 +1,50 @@
+description = 'Spring State Machine Lock Common'
+
+project('spring-statemachine-lock-shedlock') {
+	description = 'Spring State Machine Lock - ShedLock'
+	dependencies {
+		compile project(':spring-statemachine-lock-common')
+		compile group: 'net.javacrumbs.shedlock', name: 'shedlock-core', version: '3.0.0'
+		compile 'org.springframework:spring-messaging'
+
+		testCompile(project(':spring-statemachine-test')) { dep ->
+			exclude group: 'junit', module: 'junit'
+			exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
+		}
+		testRuntime 'redis.clients:jedis'
+		testCompile('org.mockito:mockito-core') { dep ->
+			exclude group: 'org.hamcrest'
+		}
+		testCompile group: 'net.javacrumbs.shedlock', name: 'shedlock-provider-redis-spring', version: '3.0.0'
+		testCompile 'org.springframework:spring-test'
+		testCompile 'org.hamcrest:hamcrest-core'
+		testCompile 'org.hamcrest:hamcrest-library'
+		testCompile("org.junit.jupiter:junit-jupiter-api")
+		testCompile("org.junit.jupiter:junit-jupiter-engine")
+		testRuntime 'org.apache.logging.log4j:log4j-core'
+	}
+}
+
+project('spring-statemachine-lock-redisson') {
+	description = 'Spring State Machine Lock - Redisson'
+	dependencies {
+		compile project(':spring-statemachine-lock-common')
+		compile 'org.redisson:redisson:3.11.4'
+		compile 'org.springframework:spring-messaging'
+
+		testCompile(project(':spring-statemachine-test')) { dep ->
+			exclude group: 'junit', module: 'junit'
+			exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
+		}
+		testRuntime 'redis.clients:jedis'
+		testCompile('org.mockito:mockito-core') { dep ->
+			exclude group: 'org.hamcrest'
+		}
+		testCompile 'org.springframework:spring-test'
+		testCompile 'org.hamcrest:hamcrest-core'
+		testCompile 'org.hamcrest:hamcrest-library'
+		testCompile("org.junit.jupiter:junit-jupiter-api")
+		testCompile("org.junit.jupiter:junit-jupiter-engine")
+		testRuntime 'org.apache.logging.log4j:log4j-core'
+	}
+}

spring-statemachine-lock/redisson/src/main/java/org/springframework/statemachine/lock/redisson/RedissonLockService.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2015-2019 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.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.statemachine.lock.redisson;
+
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.lock.LockService;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This implementation uses Redisson (https://github.com/redisson/redisson) in order to handle distributed locks on state machines.
+ * More info on https://carlosbecker.com/posts/distributed-locks-redis/ and https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers
+ *
+ * @param <S> the type of state
+ * @param <E> the type of event
+ */
+public class RedissonLockService<S, E> implements LockService<S, E> {
+
+    private final static Logger log = LoggerFactory.getLogger(RedissonLockService.class);
+
+    protected final static String LOCK_PREFIX = "lock:";
+
+    private RedissonClient redissonClient;
+    private final Map<String, RLock> locks;
+
+    public RedissonLockService(RedissonClient redissonClient) {
+        this.redissonClient = redissonClient;
+        this.locks = new ConcurrentHashMap<>();
+    }
+
+    @Override
+    public boolean lock(StateMachine<S, E> stateMachine, int lockAtMostUntil) {
+        String id = buildLockId(stateMachine);
+        RLock lock = this.redissonClient.getLock(id);
+        boolean result = false;
+        try {
+            log.trace("Lock acquired for state machine with id {}, Rlock: {}", id, lock);
+            result = lock.tryLock(0, lockAtMostUntil, TimeUnit.SECONDS);
+            this.locks.put(id, lock);
+        } catch (InterruptedException e) {
+            log.warn("Cannot acquire lock for state machine with id {}", id);
+        }
+        return result;
+    }
+
+    @Override
+    public void unLock(StateMachine<S, E> stateMachine) {
+        String id = buildLockId(stateMachine);
+        RLock lock = locks.remove(id);
+        if (lock != null) {
+            log.trace("Starting unlock on {}", id);
+            lock.unlock();
+        }
+    }
+
+    private String buildLockId(StateMachine<S, E> stateMachine) {
+        return LOCK_PREFIX + stateMachine.getId();
+    }
+
+}

spring-statemachine-lock/redisson/src/test/java/org/springframework/statemachine/lock/redisson/RedissonLockServiceTest.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2016 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.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.statemachine.lock.redisson;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.statemachine.StateMachine;
+import org.springframework.statemachine.StateMachineEventResult;
+import org.springframework.statemachine.config.EnableStateMachineFactory;
+import org.springframework.statemachine.config.StateMachineConfigurerAdapter;
+import org.springframework.statemachine.config.StateMachineFactory;
+import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineStateConfigurer;
+import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer;
+import org.springframework.statemachine.lock.LockService;
+import org.springframework.statemachine.lock.LockStateMachineGuard;
+import org.springframework.statemachine.lock.LockStateMachineListener;
+import reactor.core.publisher.Mono;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.statemachine.lock.redisson.RedissonLockService.LOCK_PREFIX;
+
+public class RedissonLockServiceTest {
+
+    private AnnotationConfigApplicationContext context;
+    private RedissonClient connection;
+
+    @BeforeEach
+    public void setup() {
+        context = buildContext();
+        context.register(RedissonConfig.class, Config1.class);
+        context.refresh();
+        connection = context.getBean(RedissonClient.class);
+        connection.getKeys().flushall();
+    }
+
+    @AfterEach
+    public void clean() {
+        if (context != null) {
+            context.close();
+        }
+    }
+
+    protected AnnotationConfigApplicationContext buildContext() {
+        return new AnnotationConfigApplicationContext();
+    }
+
+    @Test
+    public void testLockOnLockedMachine() {
+        LockService<String, String> lockService = context.getBean(LockService.class);
+        StateMachine<String, String> stateMachine = getStateMachine();
+
+        new Thread(() -> {
+            boolean lock = lockService.lock(stateMachine, 120);
+            assertThat(lock).isTrue();
+        }).start();
+
+
+        stateMachine.sendEvent(Mono.just(buildE1Event())).blockLast();
+        assertThat(stateMachine.getState().getId()).isEqualTo("S1");
+        assertThat(connection.getBucket(LOCK_PREFIX + stateMachine.getId()).isExists()).isTrue();
+    }
+
+    @Test
+    public void testLock() {
+        StateMachine<String, String> stateMachine = getStateMachine();
+        assertThat(stateMachine.getState().getId()).isEqualTo("S1");
+        StateMachineEventResult<String, String> lockResult = stateMachine.sendEvent(Mono.just(buildE1Event())).blockLast();
+        assertThat(lockResult).isNotNull();
+        assertThat(lockResult.getResultType()).isEqualTo(StateMachineEventResult.ResultType.ACCEPTED);
+        assertThat(stateMachine.getState().getId()).isEqualTo("S2");
+        assertThat(connection.getBucket(LOCK_PREFIX + stateMachine.getId()).isExists()).isFalse();
+    }
+
+    @Test
+    public void testLockInSameState() {
+        StateMachine<String, String> stateMachine = getStateMachine();
+        assertThat(stateMachine.getState().getId()).isEqualTo("S1");
+        StateMachineEventResult<String, String> lockResult = stateMachine.sendEvent(Mono.just(buildE1Event())).blockLast();
+
+        assertThat(lockResult).isNotNull();
+        assertThat(stateMachine.getState().getId()).isEqualTo("S2");
+        assertThat(connection.getBucket(LOCK_PREFIX + stateMachine.getId()).isExists()).isFalse();
+
+        StateMachineEventResult<String, String> lockResultAfter = stateMachine.sendEvent(Mono.just(buildE1Event())).blockLast();
+        assertThat(lockResultAfter).isNotNull();
+        assertThat(stateMachine.getState().getId()).isEqualTo("S2");
+        assertThat(lockResult.getResultType()).isEqualTo(StateMachineEventResult.ResultType.ACCEPTED);
+        assertThat(connection.getBucket(LOCK_PREFIX + stateMachine.getId()).isExists()).isFalse();
+    }
+
+    private StateMachine<String, String> getStateMachine() {
+        StateMachineFactory<String, String> factory = context.getBean(StateMachineFactory.class);
+        StateMachine<String, String> stateMachine = factory.getStateMachine("testId");
+        return stateMachine;
+    }
+
+    private Message<String> buildE1Event() {
+        return MessageBuilder
+                .withPayload("E1")
+                .build();
+    }
+
+    @Configuration
+    protected static class RedissonConfig {
+
+        @Bean
+        public RedissonClient redissonClient() {
+            //default address is localhost:6379
+            return Redisson.create();
+        }
+
+        @Bean
+        public LockService lockService(RedissonClient redissonClient) {
+            return new RedissonLockService(redissonClient);
+        }
+
+    }
+
+
+    @Configuration
+    @EnableStateMachineFactory
+    static class Config1 extends StateMachineConfigurerAdapter<String, String> {
+
+        @Autowired
+        private LockService<String, String> lockService;
+
+        @Override
+        public void configure(StateMachineConfigurationConfigurer<String, String> config) throws Exception {
+            config
+                    .withConfiguration()
+                    .listener(new LockStateMachineListener<>(lockService))
+                    .autoStartup(true);
+        }
+
+        @Override
+        public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
+            states
+                    .withStates()
+                    .initial("S1")
+                    .state("S2")
+                    .state("S3");
+        }
+
+        @Override
+        public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
+            transitions
+                    .withExternal()
+                    .guard(new LockStateMachineGuard<>(lockService, 120))
+                    .source("S1").target("S2")
+                    .event("E1")
+                    .action(stateContext -> System.out.println("Action on " + stateContext.getStateMachine().getId()))
+                    .and()
+                    .withExternal()
+                    .source("S2").target("S3")
+                    .event("E2")
+                    .and()
+                    .withExternal()
+                    .source("S3").target("S1")
+                    .event("E3");
+        }
+
+    }
+
+}