diff --git a/src/main/java/it/inaf/oats/vospace/AsyncTransferService.java b/src/main/java/it/inaf/oats/vospace/AsyncTransferService.java index 4819aa041c77eea7b515f7b720db8ba26c2d3312..e0fce531a4700d9d695cabc5a277be7aa1ab0d1e 100644 --- a/src/main/java/it/inaf/oats/vospace/AsyncTransferService.java +++ b/src/main/java/it/inaf/oats/vospace/AsyncTransferService.java @@ -17,6 +17,8 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.JedisPoolConfig; @Service public class AsyncTransferService { @@ -25,15 +27,22 @@ public class AsyncTransferService { private static final ObjectMapper MAPPER = new ObjectMapper(); - // WARNING: this instance is not thread safe - private final Jedis jedis; + private final JedisPool jedisPool; public AsyncTransferService(@Value("${spring.redis.host}") String redisHost, @Value("${spring.redis.port}") int redisPort) { - jedis = new Jedis(redisHost, redisPort); + + // To avoid receiving a JedisConnectionException when redis server is + // restarted a JedisPool with proper configuration is used. + JedisPoolConfig poolConfig = new JedisPoolConfig(); + + poolConfig.setTestOnBorrow(true); // sends a ping request when asking for the resource + poolConfig.setTestWhileIdle(true); // sends periodic pings to idle resources in the pool + + jedisPool = new JedisPool(poolConfig, redisHost, redisPort); } - public synchronized JobSummary startJob(JobSummary job) { - try { + public JobSummary startJob(JobSummary job) { + try (Jedis jedis = jedisPool.getResource()) { String requestId = job.getJobId(); diff --git a/src/test/java/it/inaf/oats/vospace/AsyncTransferServiceTest.java b/src/test/java/it/inaf/oats/vospace/AsyncTransferServiceTest.java index 75dd4e4adf7d757e5ddb0b3b784d9e3ea64c1e21..471cd2bd9ce83d448df78c725c42aaa632b2d42f 100644 --- a/src/test/java/it/inaf/oats/vospace/AsyncTransferServiceTest.java +++ b/src/test/java/it/inaf/oats/vospace/AsyncTransferServiceTest.java @@ -8,6 +8,7 @@ package it.inaf.oats.vospace; import it.inaf.oats.vospace.datamodel.Views; import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; import net.ivoa.xml.uws.v1.JobSummary; import net.ivoa.xml.vospace.v2.Param; import net.ivoa.xml.vospace.v2.Transfer; @@ -18,44 +19,61 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; +import org.mockito.Mock; import org.mockito.MockedConstruction; import org.mockito.Mockito; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.when; import org.mockito.junit.jupiter.MockitoExtension; import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; @ExtendWith(MockitoExtension.class) public class AsyncTransferServiceTest { private static final String JSON_JOB = "{\"jobId\":\"job_id\",\"runId\":null,\"ownerId\":null,\"phase\":null,\"quote\":null,\"creationTime\":null,\"startTime\":null,\"endTime\":null,\"executionDuration\":0,\"destruction\":null,\"parameters\":null,\"results\":[],\"errorSummary\":null,\"jobInfo\":{\"transfer\":{\"target\":\"vos://example.com!vospace/my-node\",\"direction\":\"pullToVoSpace\",\"view\":{\"param\":[{\"value\":\"file1.txt\",\"uri\":\"ivo://ia2.inaf.it/vospace/views#async-recall/include\"},{\"value\":\"file2.txt\",\"uri\":\"ivo://ia2.inaf.it/vospace/views#async-recall/include\"}],\"uri\":\"ivo://ia2.inaf.it/vospace/views#async-recall\",\"original\":true},\"protocols\":[],\"keepBytes\":false,\"version\":null,\"param\":[]}},\"version\":null}"; + @Mock + private Jedis mockedJedis; + @Test public void testRedisRpc() { - try (MockedConstruction<Jedis> staticMock = Mockito.mockConstruction(Jedis.class, - (mockedJedis, context) -> { - doAnswer(invocation -> { - String requestId = invocation.getArgument(1); - List<String> result = new ArrayList<>(); - result.add(requestId); - result.add(JSON_JOB); - return result; - }).when(mockedJedis).brpop(anyInt(), anyString()); - })) { - AsyncTransferService asyncTransferService = new AsyncTransferService("localhost", 6379); + + doAnswer(invocation -> { + String requestId = invocation.getArgument(1); + List<String> result = new ArrayList<>(); + result.add(requestId); + result.add(JSON_JOB); + return result; + }).when(mockedJedis).brpop(anyInt(), anyString()); + + testWithMockedJedis(asyncTransferService -> { JobSummary result = asyncTransferService.startJob(getFakeJob()); assertEquals("job_id", result.getJobId()); - } + }); } @Test public void testRedisError() { - try (MockedConstruction<Jedis> staticMock = Mockito.mockConstruction(Jedis.class)) { - AsyncTransferService asyncTransferService = new AsyncTransferService("localhost", 6379); + + when(mockedJedis.brpop(anyInt(), anyString())).thenReturn(List.of()); + + testWithMockedJedis(asyncTransferService -> { try { asyncTransferService.startJob(getFakeJob()); fail(); } catch (IllegalStateException ex) { } + }); + } + + private void testWithMockedJedis(Consumer<AsyncTransferService> testAction) { + try (MockedConstruction<JedisPool> staticMock = Mockito.mockConstruction(JedisPool.class, + (mockedJedisPool, context) -> { + when(mockedJedisPool.getResource()).thenReturn(mockedJedis); + })) { + AsyncTransferService asyncTransferService = new AsyncTransferService("localhost", 6379); + testAction.accept(asyncTransferService); } }