import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * Gets DB transaction isolation level based on ongoing cache transaction isolation.
 *
 * @param isolation Cache transaction isolation.
 * @return DB transaction isolation.
 */
private int isolationLevel(TransactionIsolation isolation) {
    switch (isolation) {
        case READ_COMMITTED:
            return TransactionDefinition.ISOLATION_READ_COMMITTED;

        case REPEATABLE_READ:
            return TransactionDefinition.ISOLATION_REPEATABLE_READ;

        case SERIALIZABLE:
            return TransactionDefinition.ISOLATION_SERIALIZABLE;

        default:
            throw new IllegalStateException(); // Will never happen.
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * Test checking all operations(exception resume) on suspended transaction are prohibited.
 *
 * @throws Exception If failed.
 */
public void testOpsProhibitedOnSuspendedTx() throws Exception {
    executeTestForAllCaches(new CI2Exc<Ignite, IgniteCache<Integer, Integer>>() {
        @Override public void applyx(Ignite ignite, final IgniteCache<Integer, Integer> cache) throws Exception {
            for (CI1Exc<Transaction> txOperation : SUSPENDED_TX_PROHIBITED_OPS) {
                for (TransactionIsolation isolation : TransactionIsolation.values()) {
                    Transaction tx = ignite.transactions().txStart(OPTIMISTIC, isolation);

                    cache.put(1, 1);

                    tx.suspend();

                    GridTestUtils.assertThrowsWithCause(txOperation, tx, IgniteException.class);

                    tx.resume();
                    tx.close();

                    assertNull(cache.get(1));
                }
            }
        }
    });
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void testGetReadThrough() throws Exception {
    super.testGetReadThrough();

    getReadThrough(false, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.READ_COMMITTED);
    getReadThrough(true, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.READ_COMMITTED);
    getReadThrough(false, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ);
    getReadThrough(true, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ);
    getReadThrough(false, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.SERIALIZABLE);
    getReadThrough(true, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.SERIALIZABLE);

    getReadThrough(false, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.READ_COMMITTED);
    getReadThrough(true, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.READ_COMMITTED);
    getReadThrough(false, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ);
    getReadThrough(true, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ);
    getReadThrough(false, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.SERIALIZABLE);
    getReadThrough(true, TransactionConcurrency.PESSIMISTIC, TransactionIsolation.SERIALIZABLE);
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception if failed.
 */
public void testKeepBinaryTxOverwrite() throws Exception {
    if (atomicityMode() != TRANSACTIONAL)
        return;

    IgniteCache<Integer, TestObject> cache = ignite(0).cache(DEFAULT_CACHE_NAME);

    cache.put(0, new TestObject(1));

    for (TransactionConcurrency conc : TransactionConcurrency.values()) {
        for (TransactionIsolation iso : TransactionIsolation.values()) {
            try (Transaction tx = ignite(0).transactions().txStart(conc, iso)) {
                cache.withKeepBinary().get(0);

                cache.invoke(0, new ObjectEntryProcessor());

                tx.commit();
            }
        }
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @param cache Cache.
 * @param c Cache operation Closure.
 * @param concurrency Transaction concurrency.
 * @param isolation Transaction isolation.
 * @param expLoadCnt Expected number of store 'load' calls.
 * @throws Exception If failed.
 */
private void checkReadThrough(IgniteCache<Object, Object> cache,
    IgniteRunnable c,
    @Nullable TransactionConcurrency concurrency,
    @Nullable TransactionIsolation isolation,
    int expLoadCnt) throws Exception {
    TestStore.loadCnt.set(0);

    Transaction tx = isolation != null ? cache.unwrap(Ignite.class).transactions().txStart(concurrency, isolation)
        : null;

    try {
        c.run();

        if (tx != null)
            tx.commit();
    }
    finally {
        if (tx != null)
            tx.close();
    }

    assertEquals(expLoadCnt, TestStore.loadCnt.get());
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    readExternalMeta(in);

    xidVer = (GridCacheVersion)in.readObject();
    invalidate = in.readBoolean();
    timeout = in.readLong();
    threadId = in.readLong();
    startTime = in.readLong();

    nodeId = U.readUuid(in);

    isolation = TransactionIsolation.fromOrdinal(in.read());
    concurrency = TransactionConcurrency.fromOrdinal(in.read());

    state = TransactionState.fromOrdinal(in.read());
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public Transaction txStart(TransactionConcurrency concurrency, TransactionIsolation isolation,
    long timeout, int txSize) {
    A.notNull(concurrency, "concurrency");
    A.notNull(isolation, "isolation");
    A.ensure(timeout >= 0, "timeout cannot be negative");
    A.ensure(txSize >= 0, "transaction size cannot be negative");

    return txStart0(
        concurrency,
        isolation,
        timeout,
        txSize,
        null
    ).proxy();
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/** {@inheritDoc} */
@Override public GridNearTxLocal txStartEx(
    GridCacheContext ctx,
    TransactionConcurrency concurrency,
    TransactionIsolation isolation)
{
    A.notNull(concurrency, "concurrency");
    A.notNull(isolation, "isolation");

    checkTransactional(ctx);

    TransactionConfiguration cfg = CU.transactionConfiguration(ctx, cctx.kernalContext().config());

    return txStart0(concurrency,
        isolation,
        cfg.getDefaultTxTimeout(),
        0,
        ctx.systemTx() ? ctx : null);
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * Tests preset eviction policy.
 *
 * @throws Exception If failed.
 */
private void checkPolicy0() throws Exception {
    for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
        txConcurrency = concurrency;

        for (TransactionIsolation isolation : TransactionIsolation.values()) {
            txIsolation = isolation;

            Ignite g = startGrids();

            IgniteCache<String, String> cache = g.cache(DEFAULT_CACHE_NAME);

            try {
                info(">>> Checking policy [txConcurrency=" + txConcurrency + ", txIsolation=" + txIsolation +
                    ", plc=" + plc + ", nearPlc=" + nearPlc + ']');

                checkExplicitTx(g, cache);

                checkImplicitTx(cache);
            }
            finally {
                stopAllGrids();
            }
        }
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @param client Ignite client.
 */
private void testNotMapped(IgniteEx client, TransactionConcurrency concurrency, TransactionIsolation isolation) {
    IgniteCache cache2 = client.cache(CACHE2);
    IgniteCache cache1 = client.cache(CACHE).withKeepBinary();

    try(Transaction tx = client.transactions().txStart(concurrency, isolation)) {

        Map<String, Integer> param = new TreeMap<>();
        param.put(TEST_KEY + 1, 1);
        param.put(TEST_KEY + 1, 3);
        param.put(TEST_KEY, 3);

        cache1.put(TEST_KEY, 3);

        cache1.putAll(param);
        cache2.putAll(param);

        tx.commit();
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @param cache Cache.
 * @param key Key.
 * @param concurrency Transaction concurrency.
 * @param isolation Transaction isolation.
 * @throws Exception If failed.
 */
private void checkReadThroughGetAndInvoke(IgniteCache<Object, Object> cache,
    Object key,
    TransactionConcurrency concurrency,
    TransactionIsolation isolation) throws Exception {
    putDataInStore(Collections.singletonMap(key, key), cache.getName());

    try (Transaction tx = cache.unwrap(Ignite.class).transactions().txStart(concurrency, isolation)) {
        cache.get(key);

        Object ret = cache.invoke(key, new TestEntryProcessor());

        assertEquals(key, ret);

        tx.commit();
    }

    checkValue(cache.getName(), key, (Integer)key + 1);
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testNearNodeKey() throws Exception {
    if (cacheMode() != PARTITIONED)
        return;

    if (atomicityMode() == TRANSACTIONAL) {
        for (TransactionConcurrency txConcurrency : TransactionConcurrency.values()) {
            for (TransactionIsolation txIsolation : TransactionIsolation.values()) {
                for (Operation op : Operation.values()) {
                    testNearNodeKey(txConcurrency, txIsolation, op);

                    afterTest();
                }
            }
        }
    }

    testNearNodeKey(null, null, null);
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testStoreSessionTx() throws Exception {
    testTxPut(jcache(0), null, null);

    testTxPut(ignite(0).cache(CACHE_NAME1), null, null);

    testTxRemove(null, null);

    testTxPutRemove(null, null);

    for (TransactionConcurrency concurrency : F.asList(PESSIMISTIC)) {
        for (TransactionIsolation isolation : F.asList(REPEATABLE_READ)) {
            testTxPut(jcache(0), concurrency, isolation);

            testTxRemove(concurrency, isolation);

            testTxPutRemove(concurrency, isolation);
        }
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception if failed.
 */
private void deactivateWithPendingTransaction(TransactionConcurrency concurrency,
    TransactionIsolation isolation) throws Exception {
    final Ignite ignite0 = grid(0);

    final IgniteCache<Object, Object> cache0 = ignite0.cache(CACHE_NAME);

    try (Transaction tx = ignite0.transactions().txStart(concurrency, isolation)) {
        cache0.put(1, "1");

        GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() {
            @Override public Object call() throws Exception {
                grid(0).active(false);

                return null;
            }
        }, IgniteException.class,
            "Failed to deactivate cluster (must invoke the method outside of an active transaction or lock).");
    }

    assertNull(cache0.get(1));
    assertNull(ignite0.transactions().tx());
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
private IgniteConfiguration createConfig() {
	IgniteConfiguration config = new IgniteConfiguration();
	config.setClientMode( false );
	BinaryConfiguration binaryConfiguration = new BinaryConfiguration();
	binaryConfiguration.setNameMapper( new BinaryBasicNameMapper( true ) );
	binaryConfiguration.setCompactFooter( false );		// it is necessary only for embedded collections (@ElementCollection)
	config.setBinaryConfiguration( binaryConfiguration );
	TransactionConfiguration transactionConfiguration = new TransactionConfiguration();
	transactionConfiguration.setDefaultTxConcurrency( TransactionConcurrency.OPTIMISTIC );
	transactionConfiguration.setDefaultTxIsolation( TransactionIsolation.READ_COMMITTED );
	config.setTransactionConfiguration( transactionConfiguration );

	return config;
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testPutMultipleKeysTx() throws Exception {
    for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
        for (TransactionIsolation isolation : TransactionIsolation.values()) {
            checkPutTx(true, concurrency, isolation,
                keyForNode(grid(0).localNode(), PRIMARY),
                keyForNode(grid(0).localNode(), PRIMARY),
                keyForNode(grid(0).localNode(), PRIMARY));

            checkPutTx(false, concurrency, isolation,
                keyForNode(grid(0).localNode(), PRIMARY),
                keyForNode(grid(0).localNode(), PRIMARY),
                keyForNode(grid(0).localNode(), PRIMARY));

            if (gridCount() > 1) {
                checkPutTx(true, concurrency, isolation,
                    keyForNode(grid(1).localNode(), PRIMARY),
                    keyForNode(grid(1).localNode(), PRIMARY),
                    keyForNode(grid(1).localNode(), PRIMARY));

                checkPutTx(false, concurrency, isolation,
                    keyForNode(grid(1).localNode(), PRIMARY),
                    keyForNode(grid(1).localNode(), PRIMARY),
                    keyForNode(grid(1).localNode(), PRIMARY));
            }
        }
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testTransformResourceInjection() throws Exception {
    Ignite grid = grid(0);

    IgniteCache<String, Integer> cache = grid.createCache(cacheConfiguration(ATOMIC));

    try {
        doTransformResourceInjection(cache);
    }
    finally {
        cache.destroy();
    }

    cache = grid.createCache(cacheConfiguration(TRANSACTIONAL));

    try {
        doTransformResourceInjection(cache);

        for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
            for (TransactionIsolation isolation : TransactionIsolation.values()) {
                IgniteTransactions txs = grid.transactions();

                try (Transaction tx = txs.txStart(concurrency, isolation)) {
                    doTransformResourceInjection(cache);

                    tx.commit();
                }
            }
        }
    }
    finally {
        cache.destroy();
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @param ccfg Cache configuration.
 * @throws Exception If failed.
 */
private void checkBackupConsistency(CacheConfiguration<Integer, Integer> ccfg, TransactionConcurrency txConcurrency,
    TransactionIsolation txIsolation) throws Exception {
    IgniteCache<Integer, Integer> cache = grid(0).getOrCreateCache(ccfg);

    int nodeIdx = ThreadLocalRandom.current().nextInt(NODES);

    try {
        for (int i = 0; i < ITERATION_CNT; i++) {
            log.info("Iteration: " + i);

            // Put data in one transaction.
            try (Transaction tx = grid(nodeIdx).transactions().txStart(txConcurrency, txIsolation)) {
                for (int key = 0; key < KEYS; key++)
                    cache.put(key, key);

                tx.commit();
            }

            int missCnt = 0;

            // Try to load data from another transaction.
            try (Transaction tx = grid(nodeIdx).transactions().txStart(txConcurrency, txIsolation)) {
                for (int key = 0; key < KEYS; key++)
                    if (cache.get(key) == null)
                        ++missCnt;

                tx.commit();
            }

            assertEquals("Failed. Found missing get()", 0, missCnt);
        }
    }
    finally {
        grid(0).destroyCache(ccfg.getName());
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testContainsInTx() throws Exception {
    if (atomicityMode() == TRANSACTIONAL) {
        String key = "1";

        for (int i = 0; i < gridCount(); i++)
            assertFalse("Invalid result on grid: " + i, jcache(i).containsKey(key));

        IgniteCache<String, Integer> cache = jcache(0);

        for (TransactionConcurrency conc : TransactionConcurrency.values()) {
            for (TransactionIsolation iso : TransactionIsolation.values()) {
                try (Transaction tx = grid(0).transactions().txStart(conc, iso)) {
                    assertFalse("Invalid result on grid inside tx", cache.containsKey(key));

                    assertFalse("Key was enlisted to transaction: " + tx, txContainsKey(tx, key));

                    cache.put(key, 1);

                    assertTrue("Invalid result on grid inside tx", cache.containsKey(key));

                    // Do not commit.
                }

                for (int i = 0; i < gridCount(); i++)
                    assertFalse("Invalid result on grid: " + i, jcache(i).containsKey(key));
            }
        }
    }
}
 

import org.apache.ignite.transactions.TransactionIsolation; //导入依赖的package包/类
/** */
public void configure(Ignite ignite, IgniteCache<Integer, Person> cache, TransactionConcurrency concurrency,
    TransactionIsolation isolation) {
    this.ignite = ignite;
    this.cache = cache;
    this.concurrency = concurrency;
    this.isolation = isolation;
}