Java 类org.springframework.transaction.support.TransactionCallback 实例源码

@Override
public EasyTransResult invoke(EasyTransFilterChain filterChain, Map<String, Object> header,
        EasyTransRequest<?, ?> request) {

    Integer pTrxStatus = MetaDataFilter.getMetaData(EasytransConstant.CallHeadKeys.PARENT_TRANSACTION_STATUS);
    if(!pTrxStatus.equals(com.yiqiniu.easytrans.datasource.TransStatusLogger.TransactionStatus.UNKNOWN)){
        // start transaction to update 
        PlatformTransactionManager transactionManager = getTransactionManager(filterChain, request);
        TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager,
                new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRED));
        TransactionId pTrxId = MetaDataFilter.getMetaData(EasytransConstant.CallHeadKeys.PARENT_TRX_ID_KEY);
        transactionTemplate.execute(new TransactionCallback<Object>() {
            @Override
            public Object doInTransaction(TransactionStatus status) {
                TransactionId trxId = pTrxId;
                transStatusLogger.updateExecuteFlagForSlaveTrx(trxId, request, pTrxStatus);
                return null;
            }
        });
        boolean commited = pTrxStatus.equals(com.yiqiniu.easytrans.datasource.TransStatusLogger.TransactionStatus.COMMITTED);
        //may be concurrent,but it's ok
        easyTransSynchronizer.cascadeExecuteCachedTransaction(pTrxId, commited);
    }

    return filterChain.invokeFilterChain(header, request);
}

Ticket getTicketInTransaction(final String ticketId) {
    return new TransactionTemplate(txManager).execute(new TransactionCallback<Ticket>() {
        @Override
        public Ticket doInTransaction(final TransactionStatus status) {
            return jpaTicketRegistry.getTicket(ticketId);
        }
    });
}

@Override
public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
    return new TransactionTemplate(txManager).execute(new TransactionCallback<Object>() {
        @Override
        public Object doInTransaction(final TransactionStatus status) {
            try {
                final Object result = method.invoke(jpaLock, args);
                jpaLock.entityManager.flush();
                logger.debug("Performed {} on {}", method.getName(), jpaLock);
                return result;
                // Force result of transaction to database
            } catch (final Exception e) {
                throw new RuntimeException("Transactional method invocation failed.", e);
            }
        }
    });
}

/**
* {@inheritDoc}
*/
   @Override
   public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
       return new TransactionTemplate(txManager).execute(new TransactionCallback<Object>() {
           public Object doInTransaction(final TransactionStatus status) {
               try {
                   final Object result = method.invoke(jpaLock, args);
                   jpaLock.entityManager.flush();
                   logger.debug("Performed {} on {}", method.getName(), jpaLock);
                   return result;
                   // Force result of transaction to database
               } catch (final Exception e) {
                   throw new RuntimeException("Transactional method invocation failed.", e);
               }
           }
       });
   }

/** {@inheritDoc} */
@Override
public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
    return new TransactionTemplate(txManager).execute(new TransactionCallback<Object>() {
        public Object doInTransaction(final TransactionStatus status) {
            try {
                final Object result = method.invoke(jpaLock, args);
                jpaLock.entityManager.flush();
                logger.debug("Performed {} on {}", method.getName(), jpaLock);
                return result;
                // Force result of transaction to database
            } catch (final Exception e) {
                throw new RuntimeException("Transactional method invocation failed.", e);
            }
        }
    });
}

@Override
public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
    return new TransactionTemplate(txManager).execute(new TransactionCallback<Object>() {
        @Override
        public Object doInTransaction(final TransactionStatus status) {
            try {
                final Object result = method.invoke(jpaLock, args);
                jpaLock.entityManager.flush();
                logger.debug("Performed {} on {}", method.getName(), jpaLock);
                return result;
                // Force result of transaction to database
            } catch (final Exception e) {
                throw new RuntimeException("Transactional method invocation failed.", e);
            }
        }
    });
}

@Test
public void testRwRouteWithTransaction2() {

    mapper.findByStatus((short) 1);

    transactionTemplate.execute(new TransactionCallback<Void>() {
        @Override
        public Void doInTransaction(TransactionStatus status) {

            mapper.findByStatus((short) 2);

            UserEntity entity = new UserEntity();
            entity.setCreatedAt(new Date());
            entity.setEmail(RandomStringUtils.random(6, true, true) + "@163.com");
            entity.setMobile("13800" + RandomUtils.nextLong(100000, 999999));
            entity.setType((short) 1);
            entity.setStatus((short) 1);
            mapper.insert(entity);

            mapper.findByStatus((short) 2);

            return null;
        }
    });
    System.out.println();
}

public synchronized void run(boolean useIndex) {
    if (useIndex) {
        jdbcTemplate.query(sql, setter, rowHandler);
    } else {

        long latest = latestFetcher.retrieveLatestSer();

        transactionTemplate.execute(new TransactionCallback<Object>() {
            @Override
            public Object doInTransaction(TransactionStatus status) {

                jdbcTemplate.execute("SET LOCAL enable_bitmapscan=0;");
                jdbcTemplate.query(sql, setter, rowHandler);
                return null;
            }
        });

        // shift to max(retrievedLatestSer, and ser as updated in
        // rowHandler)
        serialToContinueFrom.set(Math.max(serialToContinueFrom.get(), latest));
    }
}

@Override
    public Book searchBook(long ISBN) {
        // TODO Auto-generated method stub
         transactionTemplate.setReadOnly(true);
        return transactionTemplate.execute(new TransactionCallback<Book>() {

            @Override
            public Book doInTransaction(TransactionStatus status) {
                // TODO Auto-generated method stub
//              Book book = bookDAO.serachBook(ISBN);
                Book book=new Book();
                book.setISBN(ISBN);
                bookDAO.addBook(book);
                return book;
            }

        });

    }

@Override
public boolean updateBook(long ISBN, int price) {
    // TODO Auto-generated method stub
    if (searchBook(ISBN).getISBN() == ISBN) {
        transactionTemplate.setReadOnly(false);
    return transactionTemplate.execute(new TransactionCallback<Boolean>() {

        @Override
        public Boolean doInTransaction(TransactionStatus status) {
            // TODO Auto-generated method stub
            try {

                    int rows = bookDAO.updateBook(ISBN, price);
                    if (rows > 0)
                        return true;

            } catch (Exception exception) {
                status.setRollbackOnly();
            }
            return false;

        }
    });
    }
    return false;
}

@Override
public boolean deleteBook(long ISBN) {
    // TODO Auto-generated method stub
    if (searchBook(ISBN).getISBN() == ISBN) {
    transactionTemplate.setReadOnly(false); 
            return transactionTemplate.execute(new TransactionCallback<Boolean>() {

        @Override
        public Boolean doInTransaction(TransactionStatus status) {
            // TODO Auto-generated method stub
            try {
                    boolean result = bookDAO.deleteBook(ISBN);
                    return result;

            } catch (Exception exception) {
                status.setRollbackOnly();
            }
            return false;

        }
    });
    }
    return false;
}

/**
 * 批处理更新操作
 * 
 * @param dbInfoId
 * @param sqls
 * @return 返回更新的列的数量
 * @throws Exception
 */
public int[] updateSql(String dbInfoId, String[] sqls) throws Exception {
    final String[] fsqls = this.getFormatArrays(sqls);
    if (log.isDebugEnabled()) {
        for (String s : fsqls) {
            log.debug(s);
        }
    }
    DataSource ds = getDataSourceByDbInfoId(dbInfoId);
    final TransactionTemplate transactionTemplate = SpringJdbcUtils.getTransactionTemplate(ds);
    return transactionTemplate.execute(new TransactionCallback<int[]>() {
        public int[] doInTransaction(TransactionStatus status) {
            JdbcTemplate jdbcTemplate = SpringJdbcUtils.getJdbcTemplate(transactionTemplate);
            int[] i = jdbcTemplate.batchUpdate(fsqls);
            return i;
        }
    });
}

public boolean writeIntoFile(final File file) {
    return transactionHelper.runInCurrentThread(new TransactionCallback<Boolean>() {

        @Override
        public Boolean doInTransaction(final TransactionStatus status) {
            status.setRollbackOnly();
            try {
                final Image image = imageService.load(id);
                final InputStream in = new ByteArrayInputStream(isThumbnail ? image.getThumbnail() : image.getImage());
                file.getParentFile().mkdirs();
                IOUtils.copy(in, new FileOutputStream(file));
                file.setLastModified(image.getLastModified().getTimeInMillis());
                return true;
            } catch (final Exception e) {
                // Ignore
            }
            return false;
        }
    });
}

@Test
public void testTransactionCommitWithSharedEntityManager() {
    given(manager.getTransaction()).willReturn(tx);

    tt.execute(new TransactionCallback() {
        @Override
        public Object doInTransaction(TransactionStatus status) {
            bean.sharedEntityManager.flush();
            return null;
        }
    });

    verify(tx).commit();
    verify(manager).flush();
    verify(manager).close();
}

@Test
public void testTransactionCommitWithSharedEntityManagerAndPropagationSupports() {
    given(manager.isOpen()).willReturn(true);

    tt.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);

    tt.execute(new TransactionCallback() {
        @Override
        public Object doInTransaction(TransactionStatus status) {
            bean.sharedEntityManager.clear();
            return null;
        }
    });

    verify(manager).clear();
    verify(manager).close();
}

@Test
public void testTransactionCommitWithExtendedEntityManager() {
    given(manager.getTransaction()).willReturn(tx);

    tt.execute(new TransactionCallback() {
        @Override
        public Object doInTransaction(TransactionStatus status) {
            bean.extendedEntityManager.flush();
            return null;
        }
    });

    verify(tx, times(2)).commit();
    verify(manager).flush();
    verify(manager).close();
}

@Test
public void testTransactionCommitWithExtendedEntityManagerAndPropagationSupports() {
    given(manager.isOpen()).willReturn(true);

    tt.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);

    tt.execute(new TransactionCallback() {
        @Override
        public Object doInTransaction(TransactionStatus status) {
            bean.extendedEntityManager.flush();
            return null;
        }
    });

    verify(manager).flush();
}

@Test
public void testTransactionCommitWithPropagationSupports() {
    given(pmf.getPersistenceManager()).willReturn(pm);

    PlatformTransactionManager tm = new JdoTransactionManager(pmf);
    TransactionTemplate tt = new TransactionTemplate(tm);
    tt.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);
    final List l = new ArrayList();
    l.add("test");
    assertTrue("Hasn't thread pm", !TransactionSynchronizationManager.hasResource(pmf));

    Object result = tt.execute(new TransactionCallback() {
        @Override
        public Object doInTransaction(TransactionStatus status) {
            assertTrue("Hasn't thread pm", !TransactionSynchronizationManager.hasResource(pmf));
            assertTrue("Is not new transaction", !status.isNewTransaction());
            PersistenceManagerFactoryUtils.getPersistenceManager(pmf, true);
            return l;
        }
    });
    assertTrue("Correct result list", result == l);

    assertTrue("Hasn't thread pm", !TransactionSynchronizationManager.hasResource(pmf));

    verify(pm, times(2)).close();
}

@Test
public void uowManagerFoundInJndi() {
    MockUOWManager manager = new MockUOWManager();
    ExpectedLookupTemplate jndiTemplate =
            new ExpectedLookupTemplate(WebSphereUowTransactionManager.DEFAULT_UOW_MANAGER_NAME, manager);
    WebSphereUowTransactionManager ptm = new WebSphereUowTransactionManager();
    ptm.setJndiTemplate(jndiTemplate);
    ptm.afterPropertiesSet();

    DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
    assertEquals("result", ptm.execute(definition, new TransactionCallback<String>() {
        @Override
        public String doInTransaction(TransactionStatus status) {
            return "result";
        }
    }));

    assertEquals(UOWManager.UOW_TYPE_GLOBAL_TRANSACTION, manager.getUOWType());
    assertFalse(manager.getJoined());
    assertFalse(manager.getRollbackOnly());
}

@ManagedOperation(description = "Does the store contain the given key")
@Override
public boolean contains(final T key) {
    // Run this in single transaction.
    Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
        public Boolean doInTransaction(TransactionStatus status) {
            int count = queryForInt(key);
            if (count == 0) {
                return Boolean.FALSE;
            } else {
                return Boolean.TRUE;
            }
        }
    });
    return rc.booleanValue();

}

@Test
public void propagationNeverFailsInCaseOfExistingTransaction() {
    MockUOWManager manager = new MockUOWManager();
    manager.setUOWStatus(UOWManager.UOW_STATUS_ACTIVE);
    WebSphereUowTransactionManager ptm = new WebSphereUowTransactionManager(manager);
    DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
    definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_NEVER);

    try {
        ptm.execute(definition, new TransactionCallback<String>() {
            @Override
            public String doInTransaction(TransactionStatus status) {
                return "result";
            }
        });
        fail("Should have thrown IllegalTransactionStateException");
    }
    catch (IllegalTransactionStateException ex) {
        // expected
    }
}

@Test
public void propagationNestedFailsInCaseOfExistingTransaction() {
    MockUOWManager manager = new MockUOWManager();
    manager.setUOWStatus(UOWManager.UOW_STATUS_ACTIVE);
    WebSphereUowTransactionManager ptm = new WebSphereUowTransactionManager(manager);
    DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
    definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_NESTED);

    try {
        ptm.execute(definition, new TransactionCallback<String>() {
            @Override
            public String doInTransaction(TransactionStatus status) {
                return "result";
            }
        });
        fail("Should have thrown NestedTransactionNotSupportedException");
    }
    catch (NestedTransactionNotSupportedException ex) {
        // expected
    }
}

/**
     * acquireLock
     * 加业务锁(悲观锁)
     *
     * @param lockpk String
     * @param pk_user String
     * @return boolean
     */
    public static boolean acquireLock(final String lockpk, final String pk_user) {
        Boolean block = (Boolean)independenceTrasaction(new TransactionCallback(){
            public Object doInTransaction(TransactionStatus status) {
                if (lockpk == null || lockpk.trim().length() == 0) {
                    throw new NullPointerException("lockpk不能为空");
                }
                List count = getJdbcTemplate().queryForList("select pk_lock from sm_lock where pk_lock='" + lockpk + "'");
                if (count.size() > 0) {
                    return Boolean.valueOf(false);
                }
                String ts = getCurTs();
                getJdbcTemplate().update("insert into sm_lock(pk_lock,pk_user,ts) values(?,?,?)",
                                         new Object[] {lockpk, pk_user, ts});
//                BusinessLockVO vo = new BusinessLockVO();
//                vo.setPk_lock(lockpk);
//                vo.setPk_user(pk_user);
//                vo.setTs(date);
//                _logger.info("lock(" + lockpk +")");
                return Boolean.valueOf(true);
            }
        });        
        return block.booleanValue();
    }

private void verifyErrorTable(final JdbcTemplate jdbcTemplate,
                              final TransactionTemplate txTemplate) {
    try {
        txTemplate.execute(new TransactionCallback<Long>() {
            @Override
            public Long doInTransaction(TransactionStatus transactionStatus) {
                jdbcTemplate.update(
                        "insert into " + properties.getErrorTable() + " (table_name, error_message, payload) values (?, ?, ?)",
                        properties.getErrorTable(), "message", "payload");
                transactionStatus.setRollbackOnly();
                return null;
            }
        });
    }
    catch (DataAccessException e) {
        throw new IllegalStateException("Invalid error table specified", e);
    }
}

@Test
public void testRwRouteWithTransaction2(){

    userMapper.findByStatus((short)1);

    transactionTemplate.execute(new TransactionCallback<Void>() {
        @Override
        public Void doInTransaction(TransactionStatus status) {

            userMapper.findByStatus((short)2);

            UserEntity entity = new UserEntity();
            entity.setCreatedAt(new Date());
            entity.setEmail(RandomStringUtils.random(6, true, true) + "@163.com");
            entity.setMobile("13800"+RandomUtils.nextLong(100000, 999999));
            entity.setType((short)1);
            entity.setStatus((short)1);
            userMapper.insert(entity);

            userMapper.findByStatus((short)2);

            return null;
        }
    });
    System.out.println();
}

@Override
public void updateUser(User user) {
    //分布式锁
    MultiRetryLock lock = new MultiRetryLock("updateuser_"+user.getId());
    lock.lock();
    try {           
        final UserEntity entity = BeanCopyUtils.copy(user, UserEntity.class);
        transactionTemplate.execute(new TransactionCallback<Void>() {

            @Override
            public Void doInTransaction(TransactionStatus status) {
                userMapper.updateByPrimaryKeySelective(entity);
                return null;
            }
        });
    } finally {
        lock.unlock();
    }
}

/**
 * Inserts a TN for a transfer with the specified trace number, for the current service client
 * @return true if the TN was inserted
 */
private boolean insertTN(final Long clientId, final String traceNumber) {
    return transactionHelper.runInNewTransaction(new TransactionCallback<Boolean>() {
        @Override
        public Boolean doInTransaction(final TransactionStatus status) {
            final TraceNumber tn = new TraceNumber();
            tn.setDate(Calendar.getInstance());
            tn.setClientId(clientId);
            tn.setTraceNumber(traceNumber);
            try {
                traceNumberDao.insert(tn);
                return true;
            } catch (DaoException e) {
                status.setRollbackOnly();
                if (ExceptionUtils.indexOfThrowable(e, DataIntegrityViolationException.class) != -1) {
                    // the unique constraint was violated - It means the trace number was already stored by a payment or by other reverse.
                    // If it was inserted by a payment then we must reverse it.
                    // If was inserted by other reverse then just ignore it.
                    return false;
                } else {
                    throw e;
                }
            }
        }
    });
}

/**
 * Deal with a single action - the transaction boundary is here. The
 * processing requires numerous calls to Case service and to write to our own
 * action table. The rollback most likely to be triggered by either failing to
 * find the Case service, or if it sends back an http error status code.
 *
 * @param action the action to deal with
 * @return The resulting ActionRequest that will be added to the outbound
 *         ActionInstruction
 */
private ActionRequest processActionRequest(final Action action) {
  log.info("processing action REQUEST actionid {} caseid {} actionplanid {}", action.getActionId(),
      action.getCaseId(), action.getActionPlanId());
  return transactionTemplate.execute(new TransactionCallback<ActionRequest>() {
    // the code in this method executes in a transactional context
    public ActionRequest doInTransaction(final TransactionStatus status) {
      ActionRequest actionRequest = null;
      // update our actions state in db
      ActionDTO.ActionEvent event = action.getActionType().getResponseRequired()
          ? ActionDTO.ActionEvent.REQUEST_DISTRIBUTED : ActionDTO.ActionEvent.REQUEST_COMPLETED;
      transitionAction(action, event);
      // create the request, filling in details by GETs from casesvc
      actionRequest = prepareActionRequest(action);
      // advise casesvc to create a corresponding caseevent for our action
      caseSvcClientService.createNewCaseEvent(action, CategoryDTO.CategoryType.ACTION_CREATED);
      return actionRequest;
    }
  });
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public <T> T execute(final CommandConfig config, final Command<T> command) {
    LOGGER.debug("Running command with propagation {}", config.getTransactionPropagation());

    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.setPropagationBehavior(getPropagation(config));

    T result = transactionTemplate.execute(new TransactionCallback<T>() {
        @Override
        public T doInTransaction(TransactionStatus status) {
            return next.execute(config, command);
        }
    });

    return result;
}

@Override
public String puchase(final PurchaseContext purchaseContext) {

    return transactionTemplateBiz.execute(new TransactionCallback<String>() {

        @Override
        public String doInTransaction(TransactionStatus status) {
            String activityId = ActivityIdGenerator.getUUUId();
            try {
                DTSContext context = businessActivityControlService.start("dts-local-example");
                activityId = context.getActivityId();
                context.addArgs(PurchaseContext.class.getName(), purchaseContext);
                orderTrade.pre(context);
                yuePaymentTrade.pre(context);
                String orderId = context.getArgs("orderId", String.class);
                saveTradeFlow(orderId);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                status.setRollbackOnly();
                tradeLog.saveTradeLogIfPossible(activityId, "系统异常", e.getMessage());
            }
            return activityId;
        }
    });
}

@Test
public void shouldRollbackOnCursorManagerException() throws IOException {

    new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
        @Override
        public Void doInTransaction(TransactionStatus transactionStatus) {
            assertFalse("Transaction should be active", transactionStatus.isRollbackOnly());

            try {
                batchHandler.processBatch(new IORunnable() {
                    @Override
                    public void run() throws IOException {
                        throw new IOException("commit failed");
                    }
                });
            } catch (IOException e) {
                LOG.info("Ignoring [{}] with message [{}]", e.getClass().getName(), e.getMessage());
            }

            assertTrue("Transaction should be marked as rollback only", transactionStatus.isRollbackOnly());

            return null;
        }
    });

}

/**
 * Get user setting optional. The database call is executed in a
 * programmatic transaction.
 *
 * @param key      the user setting key.
 * @param username the username of the user.
 * @return an optional user setting value.
 */
private Optional<Serializable> getUserSettingOptional( UserSettingKey key, String username )
{
    UserCredentials userCredentials = userService.getUserCredentialsByUsername( username );

    if ( userCredentials == null )
    {
        return Optional.empty();
    }

    UserSetting setting = transactionTemplate.execute( new TransactionCallback<UserSetting>()
    {
        public UserSetting doInTransaction( TransactionStatus status )
        {
            return userSettingStore.getUserSetting( userCredentials.getUserInfo(), key.getName() );
        }
    } );

    return setting != null && setting.hasValue() ?
        Optional.of( setting.getValue() ) : Optional.empty();
}

/**
 * Finds a customer for the given username
 */
private static CustomerEntity findCustomerByName(TransactionTemplate transactionTemplate, final EntityManager entityManager, final String userName) throws Exception {
    return transactionTemplate.execute(new TransactionCallback<CustomerEntity>() {
        public CustomerEntity doInTransaction(TransactionStatus status) {
            entityManager.joinTransaction();
            List<CustomerEntity> list = entityManager.createNamedQuery("findCustomerByUsername", CustomerEntity.class).setParameter("userName", userName).getResultList();
            CustomerEntity answer;
            if (list.isEmpty()) {
                answer = new CustomerEntity();
                answer.setUserName(userName);
                LOG.info("Created a new CustomerEntity {} as no matching persisted entity found.", answer);
            } else {
                answer = list.get(0);
                LOG.info("Found a matching CustomerEntity {} having the userName {}.", answer, userName);
            }

            return answer;
        }
    });
}

@Override
public boolean contains(final Exchange exchange, final String messageId) {
    final EntityManager entityManager = getTargetEntityManager(exchange, entityManagerFactory, true, sharedEntityManager, true);

    // Run this in single transaction.
    Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
        public Boolean doInTransaction(TransactionStatus status) {
            if (isJoinTransaction()) {
                entityManager.joinTransaction();
            }

            List<?> list = query(entityManager, messageId);
            if (list.isEmpty()) {
                return Boolean.FALSE;
            } else {
                return Boolean.TRUE;
            }
        }
    });

    LOG.debug("contains {} -> {}", messageId, rc);
    return rc;
}

@ManagedOperation(description = "Clear the store")
public void clear() {
    final EntityManager entityManager = getTargetEntityManager(null, entityManagerFactory, true, sharedEntityManager, true);

    Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
        public Boolean doInTransaction(TransactionStatus status) {
            if (isJoinTransaction()) {
                entityManager.joinTransaction();
            }

            List<?> list = queryClear(entityManager);
            if (!list.isEmpty()) {
                Iterator it = list.iterator();
                while (it.hasNext()) {
                    Object item = it.next();
                    entityManager.remove(item);
                }
                entityManager.flush();
            }
            return Boolean.TRUE;
        }
    });

    LOG.debug("clear the store {}", MessageProcessed.class.getName());        
}