weld-se + infinispan transaction

[matlach]

Hi,

I'm currently using weld-se and infinispan to build a scalable application outside a j2ee container.

I'm not using any persistence unit or whatsoever just a distributed infinispan cache, backed with a store with in memory transaction.

To actually manage my transaction around my business logic, I just use the following cdi interceptor :

@InfinispanTransaction
@Interceptor
public class InfinispanTransactionInterceptor
{

    private TransactionManager transactionManager = DummyTransactionManager.getInstance();
    private boolean firstInChain;
    
    @AroundInvoke 
    public Object manageTransaction(InvocationContext ctx) throws Exception
    {
        Transaction transaction = transactionManager.getTransaction();
        if (transaction == null)
        {
            transactionManager.begin();
            firstInChain = true;
        }
        
        try
        {
            return ctx.proceed();
        }
        catch (Exception e)
        {
            transaction.setRollbackOnly();
            throw e;
        }
        finally
        {
            if (firstInChain)
            {
                if (transaction.getStatus() == Status.STATUS_ACTIVE)
                {
                    transaction.commit();
                }
                else
                {
                    transaction.rollback();
                }
            }
        }
    }
}

pretty simple stuff.

though I was wondering even if I use weld-se, since I'm managing a javax.transaction, if it would be possible to use any transactional observer, like @Observes(during=AFTER_SUCCESS), when I know the outcome of the transaction.

any idea on how I could hack this ? or wire my "managed" transaction into weld-se ?

big thanks,

[mkouba]

I guess you would have to implement your own Weld TransactionServices SPI and add it to deployment service registry (probably subclass org.jboss.weld.environment.se.Weld and override createDeployment() method). And of course you'd need a standalone JTA transaction manager to register synchronization.

[matlach]

Thanks Martin for your answer.

Here's the Weld override to add TransactionService

public class Weld extends org.jboss.weld.environment.se.Weld
{
    @Override
    protected Deployment createDeployment(ResourceLoader resourceLoader, Bootstrap bootstrap)
    {
        Deployment deployment = super.createDeployment(resourceLoader, bootstrap);
        deployment.getServices().add(TransactionServices.class, new InfinispanTransactionService());
        return deployment;
    }
}

and the TransactionService implementation

public class InfinispanTransactionService implements TransactionServices
{

    @Override
    public void cleanup()
    {
        
    }

    @Override
    public void registerSynchronization(Synchronization synchronizedObserver)
    {
        
    }

    @Override
    public boolean isTransactionActive()
    {
        try
        {
            return (getUserTransaction().getStatus() == Status.STATUS_ACTIVE);
        }
        catch (SystemException e)
        {
            return false;
        }
    }

    @Override
    public UserTransaction getUserTransaction()
    {
        return DummyTransactionManager.getUserTransaction();
    }

}

though I don't quite understand what I need to implements the registerSynchronization()  method.

When looking at the TransactionalObserverNotifier source :

    private <T> void deferNotification(final T event, final ObserverMethod<? super T> observer) {
        DeferredEventNotification<T> deferredEvent = new DeferredEventNotification<T>(event, observer);
        TransactionPhase transactionPhase = observer.getTransactionPhase();

        if (transactionPhase.equals(TransactionPhase.BEFORE_COMPLETION)) {
            transactionServices.registerSynchronization(new TransactionSynchronizedRunnable(deferredEvent, true));
        } else if (transactionPhase.equals(TransactionPhase.AFTER_COMPLETION)) {
            transactionServices.registerSynchronization(new TransactionSynchronizedRunnable(deferredEvent, false));
        } else if (transactionPhase.equals(TransactionPhase.AFTER_SUCCESS)) {
            transactionServices.registerSynchronization(new TransactionSynchronizedRunnable(deferredEvent, Status.SUCCESS));
        } else if (transactionPhase.equals(TransactionPhase.AFTER_FAILURE)) {
            transactionServices.registerSynchronization(new TransactionSynchronizedRunnable(deferredEvent, Status.FAILURE));
        }
    }

I see that my InfinispanTransactionService::registerSynchronization method will be called.

Should I store those TransactionSynchronizedRunnable in a ThreadLocal<List<Synchronization>> and manually run them by calling :

1. beforeCompletion() just before userTransaction.commit() and

2. afterCompletion() just, either, after userTransaction.commit() or userTransaction.rollback()

Am I on the right track ?

Thanks again,

[matlach]

just tested my thoughts, everythings work fine !

I'll do some code cleanup and post the complete implementation shortly.

[matlach]

here's my complete solution to implement @Observes(during=TransactionPhase....) in a weld-se + infinispan environment, feel free to add any comment if you see anything weird / wrong.

thanks again,

public class Weld extends org.jboss.weld.environment.se.Weld
{
    @Override
    protected Deployment createDeployment(ResourceLoader resourceLoader, Bootstrap bootstrap)
    {
        Deployment deployment = super.createDeployment(resourceLoader, bootstrap);
        deployment.getServices().add(TransactionServices.class, new InfinispanTransactionServices());
        return deployment;
    }
}

public class InfinispanTransactionServices implements TransactionServices
{

    @Override
    public void cleanup()
    {
        // nothing to cleanup
    }

    @Override
    public void registerSynchronization(Synchronization synchronizedObserver)
    {
        InfinispanSynchronizations.registerSynchronization(synchronizedObserver);
    }
    
    @Override
    public boolean isTransactionActive()
    {
        try
        {
            return InfinispanTransactionManager.isStatusActive();
        }
        catch (SystemException e)
        {
            return false;
        }
    }

    @Override
    public UserTransaction getUserTransaction()
    {
        return InfinispanTransactionManager.getUserTransaction();
    }

}

public class InfinispanSynchronizations
{
    private static ThreadLocal<List<Synchronization>> threadLocalSynchronization = new ThreadLocal<>();
    
    public static void registerSynchronization(Synchronization synchronization)
    {
        getThreadLocalSynchronizations().add(synchronization);
    }
    
    public static void clearSynchronizations()
    {
        threadLocalSynchronization.remove();
    }
    
    private static List<Synchronization> getThreadLocalSynchronizations()
    {
        List<Synchronization> synchronizations = threadLocalSynchronization.get();
        if (synchronizations == null)
        {
            synchronizations = new LinkedList<Synchronization>();
            threadLocalSynchronization.set(synchronizations);
        }
        return synchronizations;
    }
    
    public static void beforeCompletion()
    {
        List<Synchronization> synchronizations = threadLocalSynchronization.get();
        if (synchronizations != null)
        {
            for (Synchronization synchronization : synchronizations)
            {
                synchronization.beforeCompletion();
            }
        }
    }
    
    public static void afterCompletion(int status)
    {
        List<Synchronization> synchronizations = threadLocalSynchronization.get();
        if (synchronizations != null)
        {
            for (Synchronization synchronization : synchronizations)
            {
                synchronization.afterCompletion(status);
            }
        }
    }
    
}

public class InfinispanTransactionManager
{
    public static UserTransaction getUserTransaction()
    {
        return DummyTransactionManager.getUserTransaction();
    }
    
    public static void begin() throws NotSupportedException, SystemException
    {
        getUserTransaction().begin();
    }
    
    public static void commit() throws RollbackException, HeuristicMixedException, HeuristicRollbackException, SystemException
    {
        getUserTransaction().commit();
    }
    
    public static void rollback() throws SystemException
    {
        getUserTransaction().rollback();
    }
    
    public static void setRollbackOnly() throws SystemException
    {
        getUserTransaction().setRollbackOnly();
    }
    
    public static int getUserTransactionStatus() throws SystemException
    {
        return getUserTransaction().getStatus();
    }
    
    public static boolean isStatusNoTransaction() throws SystemException
    {
        return getUserTransactionStatus() == Status.STATUS_NO_TRANSACTION;
    }
    
    public static boolean isStatusActive() throws SystemException
    {
        return getUserTransactionStatus() == Status.STATUS_NO_TRANSACTION;
    }
    
}

@InterceptorBinding
@Inherited
@Target( { ElementType.TYPE, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
public @interface InfinispanTransactional
{

}

@InfinispanTransactional
@Interceptor
public class InfinispanTransactionalInterceptor
{
    @AroundInvoke 
    public Object manageTransaction(InvocationContext ctx) throws Exception
    {
        boolean firstInChain = false;
        if (InfinispanTransactionManager.isStatusNoTransaction())
        {
            InfinispanTransactionManager.begin();
            firstInChain = true;
        }
        
        try
        {
            return ctx.proceed();
        }
        catch (Exception e)
        {
            InfinispanTransactionManager.setRollbackOnly();
            throw e;
        }
        finally
        {
            if (firstInChain)
            {
                if (InfinispanTransactionManager.isStatusActive())
                {
                    InfinispanSynchronizations.beforeCompletion();
                    InfinispanTransactionManager.commit();
                    InfinispanSynchronizations.afterCompletion(Status.STATUS_COMMITTED);
                }
                else
                {
                    InfinispanTransactionManager.rollback();
                    InfinispanSynchronizations.afterCompletion(Status.STATUS_ROLLEDBACK);
                }
            }
            InfinispanSynchronizations.clearSynchronizations();
        }
    }
}

[alesj]

Perhaps a bit better would be to not rely on DummyTM,

but to get the TM from Infinispan's config / AdvandecCache directly -- whatever that TM impl might be.

e.g. in real prod envs I'm pretty sure the TM won't be a dummy :-)

I also think you don't need InfinispanSynchronizations class,

as there should already be a TM::registerSynchronization method which you can use,

and it then handles all the staff you manually handle via InfinispanSynchronizations.

[matlach]

Yep you're right Ales, thank you for pointing me out :

DummyTransactionManager.getInstance().getTransaction().registerSynchronization(synchronizedObserver);

For the DummyTM, for my use case in-memory transaction is just acceptable ; I can afford the risk