Prototype for asychronous Spring bean factory

AbstractAsyncFactoryBean.java

package ubic.gemma.persistence.util;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;

import javax.annotation.OverridingMethodsMustInvokeSuper;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * Base implementation of {@link AsyncFactoryBean} so that subclasses only need to implement the relevant parts of the
 * {@link FactoryBean} interface.
 * <p>
 * The implementation initializes beans using a {@link ExecutorService} which can be configured via
 * {@link #AbstractAsyncFactoryBean(ExecutorService)}. The default executor is single-threaded, which is suitable for
 * singleton beans.
 * <p>
 * Never inject {@link T} directly as this will cause a synchronous initialization of the bean. Instead, inject either a
 * {@link org.springframework.beans.factory.BeanFactory} and call {@link org.springframework.beans.factory.BeanFactory#getBean(Class)}
 * when needed or inject {@link AsyncFactoryBean} and call {@link AsyncFactoryBean#getObjectAsync()}.
 * <p>
 * This implementation handles destruction of the factory when the context is closed by dispatching a
 * {@link Future#cancel(boolean)} on any pending bean creation.
 * @param <T> type of bean that this factory provides
 * @author poirigui
 */
public abstract class AbstractAsyncFactoryBean<T> implements AsyncFactoryBean<T>, DisposableBean {

    /**
     * Executor used to initialize beans.
     */
    private final ExecutorService executor;

    /**
     * Singleton if {@link #isSingleton()} is true.
     */
    private Future<T> singletonBean;

    /**
     * Pending futures, but might also contain completed ones.
     * <p>
     * Completed futures are removed before adding new ones.
     */
    private final List<Future<T>> pendingBeans = new ArrayList<>();

    protected AbstractAsyncFactoryBean() {
        this( Executors.newSingleThreadExecutor() );
    }

    protected AbstractAsyncFactoryBean( ExecutorService executor ) {
        this.executor = executor;
    }

    @Override
    public synchronized final Future<T> getObjectAsync() {
        if ( isSingleton() && singletonBean != null ) {
            return singletonBean;
        }
        Future<T> future = executor.submit( this::getObject );
        if ( isSingleton() ) {
            singletonBean = future;
        }
        pendingBeans.removeIf( Future::isDone );
        pendingBeans.add( future );
        return future;
    }

    @Override
    @OverridingMethodsMustInvokeSuper
    public void destroy() {
        for ( Future<T> f : pendingBeans ) {
            f.cancel( true );
        }
    }
}

AsyncFactoryBean.java

package ubic.gemma.persistence.util;

import org.springframework.beans.factory.FactoryBean;

import java.util.concurrent.Future;

/**
 * Async extension for {@link FactoryBean}.
 */
public interface AsyncFactoryBean<T> extends FactoryBean<T> {

    Future<T> getObjectAsync();
}

AsyncFactoryTest.java

package ubic.gemma.core.util;

import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
import org.springframework.util.StopWatch;
import ubic.gemma.persistence.util.AbstractAsyncFactoryBean;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@ContextConfiguration
public class AsyncFactoryTest extends AbstractJUnit4SpringContextTests {
    public static class MyService {
        public MyService() throws InterruptedException {
            try {
                Thread.sleep( 1000 );
            } catch ( InterruptedException e ) {
                System.out.println( "Got interrupted!" );
                throw e;
            }
        }

    }

    public static class MyServiceFactory extends AbstractAsyncFactoryBean<MyService> {

        public MyServiceFactory() {
            super( Executors.newFixedThreadPool( 4 ) );
        }

        @Override
        public MyService getObject() throws Exception {
            return new MyService();
        }

        @Override
        public Class<?> getObjectType() {
            return MyService.class;
        }

        @Override
        public boolean isSingleton() {
            return false;
        }
    }

    @Configuration
    static class AsyncFactoryBeanTestContextConfiguration {

        @Bean
        MyServiceFactory myService() {
            return new MyServiceFactory();
        }
    }

    @Autowired
    private ConfigurableBeanFactory beanFactory;

    @Test
    public void testGetBeanAsync() {
        Future<MyService> future = beanFactory.getBean( MyServiceFactory.class ).getObjectAsync();
        Future<MyService> future2 = beanFactory.getBean( MyServiceFactory.class ).getObjectAsync();
        Assert.assertNotSame( future, future2 );
    }

    @Test
    public void testGetBeanAsyncThenWait() throws ExecutionException, InterruptedException {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        MyServiceFactory factory = beanFactory.getBean( MyServiceFactory.class );
        List<Future<MyService>> futures = new ArrayList<>();
        for ( int i = 0; i < 10; i++ ) {
            futures.add( factory.getObjectAsync() );
        }
        for ( Future<MyService> future : futures ) {
            future.get();
        }
        stopWatch.stop();
        // loading them all sequentially would take 10s
        Assert.assertTrue( stopWatch.getTotalTimeMillis() < 4000 );
    }

    @Test
    public void testGetBeanAsyncThenCancel() {
        MyServiceFactory factory = beanFactory.getBean( MyServiceFactory.class );
        List<Future<MyService>> futures = new ArrayList<>();
        for ( int i = 0; i < 10; i++ ) {
            futures.add( factory.getObjectAsync() );
        }
        factory.destroy();
        for ( Future<MyService> future : futures ) {
            Assert.assertTrue( future.isCancelled() );
        }
    }

    @Test
    public void testGetBean() {
        MyService myService = beanFactory.getBean( MyService.class );
        MyService myService1 = beanFactory.getBean( MyService.class );
        Assert.assertNotSame( myService, myService1 );
    }
}

This will be part of https://github.com/pavlidisLab/Gemma 1.30 release.