Asynchronous Processing via @Annotation + Dynamic Proxy + ExecutorService

Hello, I'd like to share an experiment based on a simple asynchronous processing class that acts as a factory and a dynamic proxy to wrap legacy implementation methods that take potentially long execution time blocking from sequential flow into asynchronous non-blocking simplest possible custom refactoring migration.

The challenges encountered here are:

[1] Legacy overloaded method signature + implementations that return a type.

[2] Provide a clean and easy meta annotation to instruct that a method is a potential task to be wrapped into it's own execution thread.

[3] Such meta annotation should also provide a way to register callback user implementations.

[4] Delegate the task to a proxy that handles all implementation details to clearly separate concerns, utilizing: thread pool, concurrent executor with runnable wrappers, etc.

Let's depict a simple legacy implementation scenario:

public interface LegacyDemoInterface {
 public T legacyProcess( final String n , final Long l );
}
// ........
public class LegacyDemoImpl implements LegacyDemoInterface {
 public String legacyProcess( final String n , final Long l ) {
   // usr impl here with long ex tm
   return "x";		
 }
}

So now imagine we could then easily annotate such method like this:

public class LegacyDemoImpl implements LegacyDemoInterface {

 @AsynchProcess(onSuccess="onAsyncSuccess",onFailure="onAsyncFailure")
 public String legacyProcess( final String n , final Long l ) {
  // usr impl here
  return "x";		
 }
 public <T> void onAsyncSuccess( final T result ) {
   System.out.println(">>onAsyncSuccess : " + result );
 }
 public void onAsyncFailure( final Throwable t ) {
  System.err.println(">>onAsyncFailure : " + t.toString());
 }
}

Providing couple callback user implementations where on the asynchronous notification will take place upon success or failure, without using Future<T> but with similar semantics such as @Async from Spring framework, in this case method 'legacyProcess(..)' returns a type(String) which will be passed as a parameter onto success callback. So let's dive into the implementation details, first the @AsynchProcess:

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface AsynchProcess {
 String onSuccess() default "onSuccess";
 String onFailure() default "onFailure";
}

Then the trick is in Dynamic Proxy Factory, note that it has anonymous class approach (on steroids :) to somewhat aprox + emulate the programming concept of 'closure' which in my words I define as: 'the ability to implement embedded functions with the capability for the inner ones to access variables outside of their scope' with corresponding 'final' parameter declarations, there's also an approach to avoid annotating interfaces as part of a proxy nature, in this case we annotated implementation class method shown above to bend some rules.

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class AsyncProcessEngine {

private static final AsyncProcessEngine instance = new AsyncProcessEngine(); //eagerly
private final Logger log = LoggerFactory.getLogger(AsyncProcessEngine.class);
private final ConcurrentHashMap<string,method> methodCache;
private final ExecutorService execService;

private AsyncProcessEngine() {
 execService = Executors.newCachedThreadPool();
 methodCache = new ConcurrentHashMap<string,method>();
 log.info("-=[ AsyncProcessor singleton ]=-");
}

public static AsyncProcessEngine getInstance() {
 return instance;
}

public static void shutdown() {
 instance.methodCache.clear();
 instance.execService.shutdown();
}

public <T> T createAsyncProxyProcess( final Object target , final Class<T> expectedType ) {
return
expectedType.cast(
  Proxy.newProxyInstance(
    target.getClass().getClassLoader() ,
    target.getClass().getInterfaces()  ,
      new InvocationHandler() {
        @Override
        public Object invoke( final Object proxy  ,
                              final Method method ,
                              final Object[] args ) throws Throwable {
       
          final Method implMethod = target.getClass().
          getMethod( method.getName() , method.getParameterTypes() );
                                                                                                
          if ( !implMethod.isAnnotationPresent( AsynchProcess.class ) ) {
               return method.invoke( target , args ); // filter
          }
        
          log.debug("{} method introspection for: {}" , target.toString() , implMethod.toGenericString() );
       
          instance.execService.submit(
             new Runnable() {
               @Override
               public void run() {
                  Method onSuccess = null;
                  Method onFailure = null;
                  try {
                                         
                     final AsynchProcess async = implMethod.getAnnotation( AsynchProcess.class );                                             
                     final String   keySuccess = target.getClass().getName() + target.hashCode() + async.onSuccess();
                     final String   keyFailure = target.getClass().getName() + target.hashCode() + async.onFailure();
                                          
                     if ( !methodCache.containsKey( keySuccess ) || !methodCache.containsKey( keyFailure ) ) {
                       try {
                         onSuccess = target.getClass().getMethod( async.onSuccess() , new Class<?>[] { Object.class    } );
                         onFailure = target.getClass().getMethod( async.onFailure() , new Class<?>[] { Throwable.class } );
                       } catch (final Exception e) { }
                       if ( onSuccess == null || onFailure == null ) {
                          return;
                       } else {
                          methodCache.put( keySuccess , onSuccess );
                          methodCache.put( keyFailure , onFailure );                
                       }
                     } else {
                       onSuccess = methodCache.get( keySuccess );
                       onFailure = methodCache.get( keyFailure );
                     }

                     ////////////////////////////////////////////////////////////////                             
                     try {                             
                        ////////////////////////////////////////////////////////////
                        //invoke target method and pass ret val into success usrImpl
                        onSuccess.invoke( target , method.invoke( target , args ) );
                        ////////////////////////////////////////////////////////////                             
                     } catch (final Throwable t ) {
                        onFailure.invoke( target , t.getCause() );
                     }
                     ////////////////////////////////////////////////////////////////

                 } catch (final Throwable e) {
                   log.error("woha@run()",e);
                 }
            }
          }
          );                    
          return null;
         }
     }     
 )
);
}
}

We could inject via DI proxied objs as shown previously here with a Spring post processor @InjectProxiedHbmTxMgr, but as usual I like keeping things lean and to have control over the behavior that I expect with minimal dependencies possible. Summary: Again, this concept could be taken to so many different levels, as demonstrated it touches very exciting topics :) specially when approaching legacy code refactoring with minimal impact possible, there are other approaches, here's one I thought of. Here's the source archive for download with a sample reference test. Alright then, till the next experiment...cheers!