Commit acfef35f authored by Sonia Zorba's avatar Sonia Zorba
Browse files

Fixed AsyncTransferService issue caused by Redis restart

parent 958e9a3e

src/main/java/it/inaf/oats/vospace/AsyncTransferService.java

+14 −5
Original line number Diff line number Diff line
@@ -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();

src/test/java/it/inaf/oats/vospace/AsyncTransferServiceTest.java

+32 −14
Original line number Diff line number Diff line
@@ -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,21 +19,26 @@ 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<>();
@@ -40,22 +46,34 @@ public class AsyncTransferServiceTest { 
            result.add(JSON_JOB);

            return result;

        }).when(mockedJedis).brpop(anyInt(), anyString());

                })) {

            AsyncTransferService asyncTransferService = new AsyncTransferService("localhost", 6379);



        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);

        }

    }