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

@Test
public void should_return_external_executor_runner() throws Exception {
    QueueConsumer queueConsumer = mock(QueueConsumer.class);
    QueueSettings settings = QueueSettings.builder().withBetweenTaskTimeout(Duration.ZERO).withNoTaskTimeout(Duration.ZERO)
            .withProcessingMode(ProcessingMode.USE_EXTERNAL_EXECUTOR).build();
    QueueLocation location = QueueLocation.builder().withTableName("testTable")
            .withQueueId(new QueueId("testQueue")).build();
    when(queueConsumer.getQueueConfig()).thenReturn(new QueueConfig(location, settings));

    QueueRunner queueRunner = QueueRunner.Factory.createQueueRunner(queueConsumer,
            new QueueDao(new QueueShardId("s1"), mock(JdbcOperations.class), mock(TransactionOperations.class)),
            mock(TaskLifecycleListener.class),
            mock(Executor.class));

    assertThat(queueRunner, CoreMatchers.instanceOf(QueueRunnerInExternalExecutor.class));

}

@Test
public void should_return_separate_transactions_runner() throws Exception {
    QueueConsumer queueConsumer = mock(QueueConsumer.class);
    QueueSettings settings = QueueSettings.builder().withBetweenTaskTimeout(Duration.ZERO).withNoTaskTimeout(Duration.ZERO)
            .withProcessingMode(ProcessingMode.SEPARATE_TRANSACTIONS).build();
    QueueLocation location = QueueLocation.builder().withTableName("testTable")
            .withQueueId(new QueueId("testQueue")).build();
    when(queueConsumer.getQueueConfig()).thenReturn(new QueueConfig(location, settings));

    QueueRunner queueRunner = QueueRunner.Factory.createQueueRunner(queueConsumer,
            new QueueDao(new QueueShardId("s1"), mock(JdbcOperations.class), mock(TransactionOperations.class)),
            mock(TaskLifecycleListener.class),
            null);

    assertThat(queueRunner, CoreMatchers.instanceOf(QueueRunnerInSeparateTransactions.class));
}

@Test
public void should_return_wrap_in_transaction_runner() throws Exception {
    QueueConsumer queueConsumer = mock(QueueConsumer.class);
    QueueSettings settings = QueueSettings.builder().withBetweenTaskTimeout(Duration.ZERO).withNoTaskTimeout(Duration.ZERO)
            .withProcessingMode(ProcessingMode.WRAP_IN_TRANSACTION).build();
    QueueLocation location = QueueLocation.builder().withTableName("testTable")
            .withQueueId(new QueueId("testQueue")).build();
    when(queueConsumer.getQueueConfig()).thenReturn(new QueueConfig(location, settings));

    QueueRunner queueRunner = QueueRunner.Factory.createQueueRunner(queueConsumer,
            new QueueDao(new QueueShardId("s1"), mock(JdbcOperations.class), mock(TransactionOperations.class)),
            mock(TaskLifecycleListener.class),
            null);

    assertThat(queueRunner, CoreMatchers.instanceOf(QueueRunnerInTransaction.class));
}

@Override
public PageSet query(String databaseName, final String sql, final Map<String, ?> paramMap, final Consumer consumer) {

    final DataSource dataSource = dataSourceDef.getDataSource(databaseName);
    PlatformTransactionManager txMgr = new DataSourceTransactionManager(dataSource);
    DefaultTransactionDefinition txDef = new DefaultTransactionDefinition();
    txDef.setReadOnly(true);

    TransactionOperations txOp = new TransactionTemplate(txMgr, txDef);
    return txOp.execute(new TransactionCallback<PageSet>() {
        @Override
        public PageSet doInTransaction(TransactionStatus status) {
            try {

                long numOfItems = extractor.extract(dataSource, sql, paramMap, consumer, new NoneLimiter());
                PageSet pageSet = paginator.paginate(0L, numOfItems, (numOfItems <= 0L ? 1L : numOfItems));

                return pageSet;
            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

@Override
public FileProcessResult handleFile(final File file, String name, String originalFilename, String contentType,
        long size, long asyncId, String... args) throws IOException {
    final DataSource dataSource = dataSourceDef.getDataSource(args[0]);
    final String sql = args[1];
    TransactionOperations txOp = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
    return txOp.execute(new TransactionCallback<FileProcessResult>() {
        @Override
        public FileProcessResult doInTransaction(TransactionStatus status) {
            try (InputStream in = new FileInputStream(file); Reader reader = new InputStreamReader(in, charset)) {

                LoadResult r = loader.load(dataSource, sql, new CsvProvider(reader, true), new NoneLimiter());

                FileProcessResult result = new FileProcessResult();
                result.setTotalCount(r.getTotalCount());
                result.setOkCount(r.getSuccessCount());
                result.setNgCount(r.getFailedCount());
                return result;

            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

@Override
public FileProcessResult handleFile(final File file, String name, String originalFilename, String contentType,
        long size, long asyncId, String... args) throws IOException {
    TransactionOperations txOp = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
    return txOp.execute(new TransactionCallback<FileProcessResult>() {
        @Override
        public FileProcessResult doInTransaction(TransactionStatus status) {
            try (InputStream in = new FileInputStream(file); Reader reader = new InputStreamReader(in, charset)) {

                LoadResult r = loader.load(dataSource, usermanImportSql, new CsvProvider(reader, true),
                        new NoneLimiter());

                FileProcessResult result = new FileProcessResult();
                result.setTotalCount(r.getTotalCount());
                result.setOkCount(r.getSuccessCount());
                result.setNgCount(r.getFailedCount());
                return result;

            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

@Override
public PageSet query(String databaseName, final String sql, final Map<String, ?> paramMap, final Consumer consumer) {

    final DataSource dataSource = dataSourceDef.getDataSource(databaseName);
    PlatformTransactionManager txMgr = new DataSourceTransactionManager(dataSource);
    DefaultTransactionDefinition txDef = new DefaultTransactionDefinition();
    txDef.setReadOnly(true);

    TransactionOperations txOp = new TransactionTemplate(txMgr, txDef);
    return txOp.execute(new TransactionCallback<PageSet>() {
        @Override
        public PageSet doInTransaction(TransactionStatus status) {
            try {

                long numOfItems = extractor.extract(dataSource, sql, paramMap, consumer, new NoneLimiter());
                PageSet pageSet = paginator.paginate(0L, numOfItems, (numOfItems <= 0L ? 1L : numOfItems));

                return pageSet;
            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

public FileProcessResult handleFile(final MultipartFile file, String dataSourceName, final String sql) {
    final DataSource dataSource = dataSourceDef.getDataSource(dataSourceName);
    TransactionOperations txOp = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
    return txOp.execute(new TransactionCallback<FileProcessResult>() {
        @Override
        public FileProcessResult doInTransaction(TransactionStatus status) {
            try (InputStream in = file.getInputStream();
                    Reader reader = new InputStreamReader(in, config.getImportCharset())) {

                LoadResult r = loader.load(dataSource, sql, new CsvProvider(reader, true), new NoneLimiter());

                FileProcessResult result = new FileProcessResult();
                result.setTotalCount(r.getTotalCount());
                result.setOkCount(r.getSuccessCount());
                result.setNgCount(r.getFailedCount());
                return result;

            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

public NsiConnectionProviderSoapClient(
    NsiHelper nsiHelper,
    AsyncSoapClient soapClient,
    Function<String, Optional<NsiProvider>> provider,
    NsiConnectionRequesterService requesterService,
    NsiV2MessageRepo messageRepo,
    TransactionOperations transactionTemplate,
    ReservationRepo reservationRepo,
    Function<String, Collection<AccessToken>> findUserAccessTokens) {
  this.nsiHelper = nsiHelper;
  this.soapClient = soapClient;
  this.findUserAccessTokens = findUserAccessTokens;
  this.findProvider = nsa -> provider.apply(nsa).orElseThrow(() -> new IllegalStateException("cannot send request to unknown provider '" + nsa + "'"));
  this.requesterService = requesterService;
  this.messageRepo = messageRepo;
  this.transactionTemplate = transactionTemplate;
  this.reservationRepo = reservationRepo;
}

@Override
public TransactionOperations getExecuteTransactionOperations() {
    return new TransactionOperations() {
        @Override
        public <T2> T2 execute(TransactionCallback<T2> action) throws TransactionException {
            return action.doInTransaction(new SimpleTransactionStatus());
        }
    };
}

@Override
public PageSet query(String databaseName, final QueryBuilder queryBuilder, final Map<String, ?> paramMap,
        final long pageNo, final long pageSz, final Consumer consumer) {

    final DataSource dataSource = dataSourceDef.getDataSource(databaseName);
    PlatformTransactionManager txMgr = new DataSourceTransactionManager(dataSource);
    DefaultTransactionDefinition txDef = new DefaultTransactionDefinition();
    txDef.setReadOnly(true);

    TransactionOperations txOp = new TransactionTemplate(txMgr, txDef);
    return txOp.execute(new TransactionCallback<PageSet>() {
        @Override
        public PageSet doInTransaction(TransactionStatus status) {
            try {

                long count = count(dataSource, queryBuilder.buildCount(), paramMap);
                PageSet pageSet = paginator.paginate(pageNo, count, pageSz);

                long numOfItems = extractor.extract(dataSource,
                        queryBuilder.build(pageSz, pageSet.getCurrent().getFrom()), paramMap, consumer,
                        new NoneLimiter());
                if (numOfItems != pageSet.getCurrent().getCount()) {
                    throw new IllegalStateException();
                }

                return pageSet;
            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

@Override
public PageSet query(String databaseName, final QueryBuilder queryBuilder, final Map<String, ?> paramMap,
        final long pageNo, final long pageSz, final Consumer consumer) {

    final DataSource dataSource = dataSourceDef.getDataSource(databaseName);
    PlatformTransactionManager txMgr = new DataSourceTransactionManager(dataSource);
    DefaultTransactionDefinition txDef = new DefaultTransactionDefinition();
    txDef.setReadOnly(true);

    TransactionOperations txOp = new TransactionTemplate(txMgr, txDef);
    return txOp.execute(new TransactionCallback<PageSet>() {
        @Override
        public PageSet doInTransaction(TransactionStatus status) {
            try {

                long count = count(dataSource, queryBuilder.buildCount(), paramMap);
                PageSet pageSet = paginator.paginate(pageNo, count, pageSz);

                long numOfItems = extractor.extract(dataSource,
                        queryBuilder.build(pageSz, pageSet.getCurrent().getFrom()), paramMap, consumer,
                        new NoneLimiter());
                if (numOfItems != pageSet.getCurrent().getCount()) {
                    throw new IllegalStateException();
                }

                return pageSet;
            } catch (IOException ex) {
                throw new IllegalStateException(ex);
            }
        }
    });
}

public TransactionalJob(Progress progress, TransactionOperations transactionOperations,
        Authentication authentication)
{
    this.progress = requireNonNull(progress);
    this.transactionOperations = requireNonNull(transactionOperations);
    this.authentication = requireNonNull(authentication);
}

private <T> T tryCallInTransaction(Job<T> job, Progress progress, TransactionOperations transactionOperations)
{
    try
    {
        return transactionOperations.execute((status) -> tryCall(job, progress));
    }
    catch (TransactionException te)
    {
        LOG.error("Transaction error while running job", te);
        progress.failed(te);
        throw te;
    }
}

@DependsOn("testShardRouter2")
@Bean
QueueDao queueDao1() {
    return new QueueDao(new QueueShardId("1"), mock(JdbcOperations.class),
            mock(TransactionOperations.class));
}

@DependsOn("queueDao1")
@Bean
QueueDao queueDao2() {
    return new QueueDao(new QueueShardId("2"), mock(JdbcOperations.class),
            mock(TransactionOperations.class));
}

@DependsOn("testShardRouter2")
@Bean
QueueDao queueDao1() {
    return new QueueDao(new QueueShardId("1"), mock(JdbcOperations.class),
            mock(TransactionOperations.class));
}

@DependsOn("queueDao1")
@Bean
QueueDao queueDao2() {
    return new QueueDao(new QueueShardId("1"), mock(JdbcOperations.class),
            mock(TransactionOperations.class));
}

public <T> void runWithTransaction(final String loggerContext, Collection<? extends Runnable> runnables,
        TransactionOperations TransactionOperations, Integer totalItems) throws ExecutionException {
    callWithTransaction(loggerContext, Collections2.transform(runnables, RUNNABLE_TO_CALLABLE),
            TransactionOperations, totalItems);
}

public TransactionWrapperCallable(TransactionOperations transactionTemplate, Callable<T> callable) {
    super();
    this.transactionTemplate = transactionTemplate;
    this.callable = callable;
}

@Override
public TransactionOperations getExecuteTransactionOperations() {
    return executor.newTransactionTemplate(
            runnable.getWriteability(), TransactionDefinition.PROPAGATION_REQUIRED
    );
}

public void setTransactionOperation(final TransactionOperations tx) {
    m_transactionOperations = tx;
}

public void setTransactionOperations(final TransactionOperations tx) {
    m_transaction = tx;
}

public void setTransactionOperations(final TransactionOperations tx) {
    m_transaction = tx;
}

@VisibleForTesting
void setTransactionOperations(TransactionOperations transactionOperations) {
  this.transactionOperations = transactionOperations;
}

@VisibleForTesting
void setTransactionOperations(TransactionOperations transactionOperations) {
  this.transactionOperations = transactionOperations;
}

/**
 * Конструктор
 *
 * @param shardId             идентификатор шарда на котором будет происходить выборка задач
 * @param jdbcTemplate        spring jdbc template
 * @param transactionTemplate spring transaction template
 */
PickTaskDao(@Nonnull QueueShardId shardId, @Nonnull JdbcOperations jdbcTemplate,
            @Nonnull TransactionOperations transactionTemplate) {
    this.shardId = requireNonNull(shardId);
    this.jdbcTemplate = new NamedParameterJdbcTemplate(requireNonNull(jdbcTemplate));
    this.transactionTemplate = requireNonNull(transactionTemplate);
}

/**
 * Executes a job in the calling thread within a transaction.
 *
 * @param job                   the {@link Job} to execute.
 * @param progress              {@link Progress} to report progress to
 * @param authentication        {@link Authentication} to run the job with
 * @param transactionOperations TransactionOperations to use for a transactional call
 * @param <T>                   Job result type
 * @return the result of the job execution
 * @throws JobExecutionException if job execution throws an exception
 * @deprecated Create a service bean with a @Transactional annotation instead
 */
@Deprecated
<T> T call(Job<T> job, Progress progress, Authentication authentication,
        TransactionOperations transactionOperations)
{
    return runWithAuthentication(authentication, () -> tryCallInTransaction(job, progress, transactionOperations));
}

/**
 * Конструктор.
 * <p>
 * Клиентский код обязан обеспечить уникальную привязку пары (jdbcTemplate, transactionTemplate) к shardId
 *
 * @param shardId             идентификатор шарда
 * @param jdbcTemplate        spring jdbc template
 * @param transactionTemplate spring transaction template
 */
public QueueDao(QueueShardId shardId, JdbcOperations jdbcTemplate, TransactionOperations transactionTemplate) {
    this.shardId = shardId;
    this.jdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
    this.transactionTemplate = transactionTemplate;
}

/**
 * Получить transaction template данного шарда
 *
 * @return transaction template данного шарда
 */
public TransactionOperations getTransactionTemplate() {
    return transactionTemplate;
}

/**
 * Конструктор
 *
 * @param jdbcTemplate        spring jdbc template
 * @param transactionTemplate spring transaction template
 */
public QueueActorDao(JdbcOperations jdbcTemplate, TransactionOperations transactionTemplate) {
    this.jdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
    this.transactionTemplate = transactionTemplate;
}

/**
 * Получить transaction template
 *
 * @return spring transaction template
 */
public TransactionOperations getTransactionTemplate() {
    return transactionTemplate;
}

/**
 * Конструктор
 *
 * @param jdbcTemplate        spring jdbc template
 * @param transactionTemplate spring transaction template
 */
public QueueStatisticsDao(JdbcOperations jdbcTemplate, TransactionOperations transactionTemplate) {
    this.jdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
    this.transactionTemplate = transactionTemplate;
}

/**
 * Получить transaction template
 *
 * @return spring transaction template
 */
public TransactionOperations getTransactionTemplate() {
    return transactionTemplate;
}

/**
 * Получить transaction template данного шарда
 *
 * @return transaction template данного шарда
 */
TransactionOperations getTransactionTemplate() {
    return transactionTemplate;
}

public abstract TransactionOperations getExecuteTransactionOperations();