You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vk...@apache.org on 2015/05/12 08:32:52 UTC
incubator-ignite git commit: # IGNITE-891 - Cache store refactoring
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-891 [created] b3d608367
# IGNITE-891 - Cache store refactoring
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/b3d60836
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/b3d60836
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/b3d60836
Branch: refs/heads/ignite-891
Commit: b3d6083673b4226e1e64f1448b24bd2a82beac91
Parents: f6012f1
Author: Valentin Kulichenko <vk...@gridgain.com>
Authored: Mon May 11 23:32:41 2015 -0700
Committer: Valentin Kulichenko <vk...@gridgain.com>
Committed: Mon May 11 23:32:41 2015 -0700
----------------------------------------------------------------------
.../store/jdbc/CacheJdbcPersonStore.java | 146 +++++--------------
.../cache/store/CacheLoadOnlyStoreAdapter.java | 6 +
.../apache/ignite/cache/store/CacheStore.java | 10 ++
.../ignite/cache/store/CacheStoreAdapter.java | 20 +++
.../store/jdbc/CacheAbstractJdbcStore.java | 6 +
.../store/tx/CacheStoreJdbcSessionListener.java | 84 +++++++++++
.../store/tx/CacheStoreSessionListener.java | 40 +++++
.../configuration/CacheConfiguration.java | 1 -
.../configuration/IgniteReflectionFactory.java | 39 ++++-
.../cache/CacheStoreBalancingWrapper.java | 6 +
.../cache/GridCacheLoaderWriterStore.java | 6 +
.../store/GridCacheStoreManagerAdapter.java | 96 +++++++-----
.../cache/store/GridCacheWriteBehindStore.java | 6 +
.../store/GridCacheBalancingStoreSelfTest.java | 6 +
.../cache/store/GridGeneratingTestStore.java | 6 +
.../IgniteCacheExpiryStoreLoadSelfTest.java | 6 +
.../GridCacheAbstractLocalStoreSelfTest.java | 6 +
...idCacheConfigurationConsistencySelfTest.java | 6 +
.../cache/GridCacheGenericTestStore.java | 6 +
.../cache/GridCacheLifecycleAwareSelfTest.java | 6 +
.../cache/GridCacheStorePutxSelfTest.java | 7 +-
.../processors/cache/GridCacheTestStore.java | 6 +
.../IgniteTxStoreExceptionAbstractSelfTest.java | 6 +
.../IgniteCrossCacheTxStoreSelfTest.java | 6 +
.../IgniteCacheStoreSessionAbstractTest.java | 6 +
...acheStoreSessionWriteBehindAbstractTest.java | 6 +
modules/spring/pom.xml | 6 +
.../tx/CacheStoreSpringSessionListener.java | 84 +++++++++++
28 files changed, 482 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
index 791f861..d930615 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datagrid/store/jdbc/CacheJdbcPersonStore.java
@@ -19,15 +19,16 @@ package org.apache.ignite.examples.datagrid.store.jdbc;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.examples.datagrid.store.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.resources.*;
-import org.jetbrains.annotations.*;
+import org.springframework.jdbc.datasource.*;
import javax.cache.*;
import javax.cache.integration.*;
+import javax.sql.*;
import java.sql.*;
-import java.util.*;
/**
* Example of {@link CacheStore} implementation that uses JDBC
@@ -35,8 +36,11 @@ import java.util.*;
*
*/
public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
- /** Transaction metadata attribute name. */
- private static final String ATTR_NAME = "SIMPLE_STORE_CONNECTION";
+ /** Database URL. */
+ private static final String DB_URL = "jdbc:h2:mem:example;DB_CLOSE_DELAY=-1";
+
+ /** Data source. */
+ private DataSource dataSrc;
/** Auto-injected store session. */
@CacheStoreSessionResource
@@ -48,6 +52,10 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
* @throws IgniteException If failed.
*/
public CacheJdbcPersonStore() throws IgniteException {
+ dataSrc = new DriverManagerDataSource(DB_URL);
+
+ setSessionListener(new CacheStoreJdbcSessionListener(dataSrc));
+
prepareDb();
}
@@ -58,11 +66,9 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
* @throws IgniteException If failed.
*/
private void prepareDb() throws IgniteException {
- try (Connection conn = openConnection(false); Statement st = conn.createStatement()) {
+ try (Connection conn = connection(); Statement st = conn.createStatement()) {
st.execute("create table if not exists PERSONS (id number unique, firstName varchar(255), " +
- "lastName varchar(255))");
-
- conn.commit();
+ "" + "lastName varchar(255))");
}
catch (SQLException e) {
throw new IgniteException("Failed to create database table.", e);
@@ -70,32 +76,11 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
}
/** {@inheritDoc} */
- @Override public void sessionEnd(boolean commit) {
- Map<String, Connection> props = ses.properties();
-
- try (Connection conn = props.remove(ATTR_NAME)) {
- if (conn != null) {
- if (commit)
- conn.commit();
- else
- conn.rollback();
- }
-
- System.out.println(">>> Transaction ended [commit=" + commit + ']');
- }
- catch (SQLException e) {
- throw new CacheWriterException("Failed to end transaction: " + ses.transaction(), e);
- }
- }
-
- /** {@inheritDoc} */
@Override public Person load(Long key) {
System.out.println(">>> Loading key: " + key);
- Connection conn = null;
-
try {
- conn = connection();
+ Connection conn = connection();
try (PreparedStatement st = conn.prepareStatement("select * from PERSONS where id=?")) {
st.setString(1, key.toString());
@@ -109,9 +94,6 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
catch (SQLException e) {
throw new CacheLoaderException("Failed to load object: " + key, e);
}
- finally {
- end(conn);
- }
return null;
}
@@ -124,10 +106,8 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
System.out.println(">>> Putting [key=" + key + ", val=" + val + ']');
- Connection conn = null;
-
try {
- conn = connection();
+ Connection conn = connection();
int updated;
@@ -157,19 +137,14 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
catch (SQLException e) {
throw new CacheLoaderException("Failed to put object [key=" + key + ", val=" + val + ']', e);
}
- finally {
- end(conn);
- }
}
/** {@inheritDoc} */
@Override public void delete(Object key) {
System.out.println(">>> Removing key: " + key);
- Connection conn = null;
-
try {
- conn = connection();
+ Connection conn = connection();
try (PreparedStatement st = conn.prepareStatement("delete from PERSONS where id=?")) {
st.setLong(1, (Long)key);
@@ -180,9 +155,6 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
catch (SQLException e) {
throw new CacheWriterException("Failed to remove object: " + key, e);
}
- finally {
- end(conn);
- }
}
/** {@inheritDoc} */
@@ -192,84 +164,32 @@ public class CacheJdbcPersonStore extends CacheStoreAdapter<Long, Person> {
final int entryCnt = (Integer)args[0];
- try (Connection conn = connection()) {
- try (PreparedStatement st = conn.prepareStatement("select * from PERSONS")) {
- try (ResultSet rs = st.executeQuery()) {
- int cnt = 0;
-
- while (cnt < entryCnt && rs.next()) {
- Person person = new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
+ try (
+ Connection conn = connection();
+ PreparedStatement st = conn.prepareStatement("select * from PERSONS");
+ ResultSet rs = st.executeQuery()
+ ) {
+ int cnt = 0;
- clo.apply(person.getId(), person);
+ while (cnt < entryCnt && rs.next()) {
+ Person person = new Person(rs.getLong(1), rs.getString(2), rs.getString(3));
- cnt++;
- }
+ clo.apply(person.getId(), person);
- System.out.println(">>> Loaded " + cnt + " values into cache.");
- }
+ cnt++;
}
+
+ System.out.println(">>> Loaded " + cnt + " values into cache.");
}
catch (SQLException e) {
throw new CacheLoaderException("Failed to load values from cache store.", e);
}
}
- /**
- * @return Connection.
- * @throws SQLException In case of error.
- */
- private Connection connection() throws SQLException {
- // If there is an ongoing transaction,
- // we must reuse the same connection.
- if (ses.isWithinTransaction()) {
- Map<Object, Object> props = ses.properties();
-
- Connection conn = (Connection)props.get(ATTR_NAME);
-
- if (conn == null) {
- conn = openConnection(false);
-
- // Store connection in session properties, so it can be accessed
- // for other operations on the same transaction.
- props.put(ATTR_NAME, conn);
- }
-
- return conn;
- }
- // Transaction can be null in case of simple load or put operation.
+ private Connection connection() throws SQLException {
+ if (ses != null && ses.isWithinTransaction())
+ return ses.<String, Connection>properties().get(CacheStoreJdbcSessionListener.CONN_KEY);
else
- return openConnection(true);
- }
-
- /**
- * Closes allocated resources depending on transaction status.
- *
- * @param conn Allocated connection.
- */
- private void end(@Nullable Connection conn) {
- if (!ses.isWithinTransaction() && conn != null) {
- // Close connection right away if there is no transaction.
- try {
- conn.close();
- }
- catch (SQLException ignored) {
- // No-op.
- }
- }
- }
-
- /**
- * Gets connection from a pool.
- *
- * @param autocommit {@code true} If connection should use autocommit mode.
- * @return Pooled connection.
- * @throws SQLException In case of error.
- */
- private Connection openConnection(boolean autocommit) throws SQLException {
- Connection conn = DriverManager.getConnection("jdbc:h2:mem:example;DB_CLOSE_DELAY=-1");
-
- conn.setAutoCommit(autocommit);
-
- return conn;
+ return dataSrc.getConnection();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/CacheLoadOnlyStoreAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheLoadOnlyStoreAdapter.java b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheLoadOnlyStoreAdapter.java
index a0cbe65..ae4bb5d 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheLoadOnlyStoreAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheLoadOnlyStoreAdapter.java
@@ -18,6 +18,7 @@
package org.apache.ignite.cache.store;
import org.apache.ignite.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.resources.*;
@@ -259,6 +260,11 @@ public abstract class CacheLoadOnlyStoreAdapter<K, V, I> implements CacheStore<K
// No-op.
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* Worker.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
index d018298..19c94d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStore.java
@@ -18,6 +18,7 @@
package org.apache.ignite.cache.store;
import org.apache.ignite.cache.store.jdbc.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.transactions.*;
import org.jetbrains.annotations.*;
@@ -94,6 +95,15 @@ public interface CacheStore<K, V> extends CacheLoader<K, V>, CacheWriter<K, V> {
* @throws CacheWriterException If commit or rollback failed. Note that commit failure in some cases
* may bring cache transaction into {@link TransactionState#UNKNOWN} which will
* consequently cause all transacted entries to be invalidated.
+ * @deprecated Deprecated in favor of {@link CacheStoreSessionListener}.
*/
+ @Deprecated
public void sessionEnd(boolean commit) throws CacheWriterException;
+
+ /**
+ * Gets session listener.
+ *
+ * @return Session listener.
+ */
+ public CacheStoreSessionListener getSessionListener();
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreAdapter.java b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreAdapter.java
index f7adf2c..0c66c86 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/CacheStoreAdapter.java
@@ -17,6 +17,7 @@
package org.apache.ignite.cache.store;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.lang.*;
import javax.cache.*;
@@ -37,6 +38,23 @@ import java.util.*;
* specific arguments.
*/
public abstract class CacheStoreAdapter<K, V> implements CacheStore<K, V> {
+ /** Transaction manager. */
+ private CacheStoreSessionListener sesLsnr;
+
+ /**
+ * Sets session listener.
+ *
+ * @param sesLsnr Session listener.
+ */
+ public void setSessionListener(CacheStoreSessionListener sesLsnr) {
+ this.sesLsnr = sesLsnr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return sesLsnr;
+ }
+
/**
* Default empty implementation. This method needs to be overridden only if
* {@link org.apache.ignite.IgniteCache#loadCache(IgniteBiPredicate, Object...)} method
@@ -87,7 +105,9 @@ public abstract class CacheStoreAdapter<K, V> implements CacheStore<K, V> {
* for all other cases this method should be overridden with custom commit/rollback logic.
*
* @param commit {@inheritDoc}
+ * @deprecated {@inheritDoc}
*/
+ @Deprecated
@Override public void sessionEnd(boolean commit) {
// No-op.
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
index 22d6d7a..3f42407 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/jdbc/CacheAbstractJdbcStore.java
@@ -21,6 +21,7 @@ import org.apache.ignite.*;
import org.apache.ignite.cache.*;
import org.apache.ignite.cache.store.*;
import org.apache.ignite.cache.store.jdbc.dialect.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.util.tostring.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -342,6 +343,11 @@ public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>,
}
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* Retrieves the value of the designated column in the current row of this <code>ResultSet</code> object and
* will convert to the requested Java data type.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreJdbcSessionListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreJdbcSessionListener.java b/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreJdbcSessionListener.java
new file mode 100644
index 0000000..3ad14ad
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreJdbcSessionListener.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.store.tx;
+
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import javax.cache.*;
+import javax.sql.*;
+import java.sql.*;
+
+/**
+ * TODO
+ */
+public class CacheStoreJdbcSessionListener implements CacheStoreSessionListener {
+ /** */
+ public static final String CONN_KEY = "__conn_";
+
+ /** */
+ private DataSource dataSrc;
+
+ public CacheStoreJdbcSessionListener(DataSource dataSrc) {
+ this.dataSrc = dataSrc;
+ }
+
+ public DataSource getDataSource() {
+ return dataSrc;
+ }
+
+ public void setDataSource(DataSource dataSrc) {
+ this.dataSrc = dataSrc;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onSessionStart(CacheStoreSession ses) {
+ assert !ses.properties().containsKey(CONN_KEY);
+
+ try {
+ Connection conn = dataSrc.getConnection();
+
+ conn.setAutoCommit(false);
+
+ ses.properties().put(CONN_KEY, conn);
+ }
+ catch (SQLException e) {
+ throw new CacheException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onSessionEnd(CacheStoreSession ses, boolean commit) {
+ Connection conn = ses.<String, Connection>properties().remove(CONN_KEY);
+
+ if (conn != null) {
+ try {
+ if (commit)
+ conn.commit();
+ else
+ conn.rollback();
+ }
+ catch (SQLException e) {
+ throw new CacheException(e);
+ }
+ finally {
+ U.closeQuiet(conn);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSessionListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSessionListener.java b/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSessionListener.java
new file mode 100644
index 0000000..2157def
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSessionListener.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.store.tx;
+
+import org.apache.ignite.cache.store.*;
+
+/**
+ * Transaction manager for persistence store.
+ */
+public interface CacheStoreSessionListener {
+ /**
+ * Called when session is started.
+ *
+ * @param ses Store session.
+ */
+ public void onSessionStart(CacheStoreSession ses);
+
+ /**
+ * Called when session is finished.
+ *
+ * @param ses Store session.
+ * @param commit Whether transaction was committed.
+ */
+ public void onSessionEnd(CacheStoreSession ses, boolean commit);
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
index df6b2ee..692f1e3 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/CacheConfiguration.java
@@ -774,7 +774,6 @@ public class CacheConfiguration<K, V> extends MutableConfiguration<K, V> {
*
* @param loadPrevVal Load previous value flag.
* @return {@code this} for chaining.
- * @return {@code this} for chaining.
*/
public CacheConfiguration setLoadPreviousValue(boolean loadPrevVal) {
this.loadPrevVal = loadPrevVal;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/configuration/IgniteReflectionFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteReflectionFactory.java b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteReflectionFactory.java
index 3222938..59d8ae7 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteReflectionFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteReflectionFactory.java
@@ -39,11 +39,18 @@ public class IgniteReflectionFactory<T> implements Factory<T> {
private volatile Class<? extends T> cls;
/** */
+ private volatile Object[] args;
+
+ /** */
private volatile Map<String, Serializable> props;
/** */
private transient T instance;
+ public static <T> Factory<T> factoryOf(Class<? extends T> cls, Object... args) {
+ return new IgniteReflectionFactory<>(cls, args);
+ }
+
/**
*
*/
@@ -54,19 +61,22 @@ public class IgniteReflectionFactory<T> implements Factory<T> {
/**
* @param cls Component class.
* @param singleton Singleton flag.
+ * @param args Constructor arguments.
*/
- public IgniteReflectionFactory(Class<? extends T> cls, boolean singleton) {
+ public IgniteReflectionFactory(Class<? extends T> cls, boolean singleton, Object... args) {
this.cls = cls;
this.singleton = singleton;
+ this.args = args;
}
/**
* Creates non-singleton component factory.
*
* @param cls Component class.
+ * @param args Constructor arguments.
*/
- public IgniteReflectionFactory(Class<? extends T> cls) {
- this(cls, false);
+ public IgniteReflectionFactory(Class<? extends T> cls, Object... args) {
+ this(cls, false, args);
}
/**
@@ -128,18 +138,37 @@ public class IgniteReflectionFactory<T> implements Factory<T> {
/**
* @return Initialized instance.
*/
+ @SuppressWarnings("unchecked")
private T createInstance() {
if (cls == null)
throw new IllegalStateException("Failed to create object (object type is not set).");
try {
- T obj = cls.newInstance();
+ T obj = null;
+
+ if (args == null || args.length == 0)
+ obj = cls.newInstance();
+ else {
+ for (Constructor<?> cons : cls.getDeclaredConstructors()) {
+ try {
+ obj = (T)cons.newInstance(args);
+
+ break;
+ }
+ catch (IllegalArgumentException | ClassCastException ignored) {
+ // No-op.
+ }
+ }
+
+ if (obj == null)
+ throw new IllegalArgumentException("");
+ }
injectProperties(obj);
return obj;
}
- catch (InstantiationException | IllegalAccessException e) {
+ catch (InvocationTargetException | InstantiationException | IllegalAccessException e) {
throw new CacheException("Failed to instantiate factory object: " + cls.getName(), e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java
index 7bbf08b..8f6c0ff 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStoreBalancingWrapper.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.util.future.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.lang.*;
@@ -222,6 +223,11 @@ public class CacheStoreBalancingWrapper<K, V> implements CacheStore<K, V> {
delegate.sessionEnd(commit);
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return delegate.getSessionListener();
+ }
+
/**
*
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheLoaderWriterStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheLoaderWriterStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheLoaderWriterStore.java
index 0413853..1c9c07f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheLoaderWriterStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheLoaderWriterStore.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.lifecycle.*;
import org.jetbrains.annotations.*;
@@ -140,4 +141,9 @@ class GridCacheLoaderWriterStore<K, V> implements CacheStore<K, V>, LifecycleAwa
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
index f9a966c..a090f8a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.store;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
@@ -59,6 +60,9 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
private ThreadLocal<SessionData> sesHolder;
/** */
+ private ThreadLocalSession locSes;
+
+ /** */
private boolean locStore;
/** */
@@ -84,14 +88,15 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sesHolder0 = ((Map<CacheStore, ThreadLocal>)sesHolders).get(cfgStore);
if (sesHolder0 == null) {
- ThreadLocalSession locSes = new ThreadLocalSession();
+ sesHolder0 = new ThreadLocal<>();
- if (ctx.resource().injectStoreSession(cfgStore, locSes)) {
- sesHolder0 = locSes.sesHolder;
+ locSes = new ThreadLocalSession(sesHolder0);
+ if (ctx.resource().injectStoreSession(cfgStore, locSes))
sesHolders.put(cfgStore, sesHolder0);
- }
}
+ else
+ locSes = new ThreadLocalSession(sesHolder0);
}
sesHolder = sesHolder0;
@@ -215,14 +220,14 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
Object val = null;
try {
val = singleThreadGate.load(storeKey);
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -234,7 +239,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(new CacheLoaderException(e));
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -349,7 +354,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
IgniteBiInClosure<Object, Object> c = new CI2<Object, Object>() {
@@ -380,7 +385,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
else
singleThreadGate.loadAll(keys0, c);
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -392,7 +397,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(new CacheLoaderException(e));
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -408,7 +413,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(null);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
store.loadCache(new IgniteBiInClosure<Object, Object>() {
@@ -431,7 +436,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
}
}, args);
- thewEx = false;
+ threwEx = false;
}
catch (CacheLoaderException e) {
throw new IgniteCheckedException(e);
@@ -440,7 +445,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(new CacheLoaderException(e));
}
finally {
- sessionEnd0(null, thewEx);
+ sessionEnd0(null, threwEx);
}
if (log.isDebugEnabled())
@@ -473,12 +478,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
store.write(new CacheEntryImpl<>(key, locStore ? F.t(val, ver) : val));
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -490,7 +495,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(new CacheWriterException(e));
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -522,12 +527,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
store.writeAll(entries);
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -548,7 +553,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(e);
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -576,12 +581,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
store.delete(key);
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -593,7 +598,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(new CacheWriterException(e));
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -625,12 +630,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
- boolean thewEx = true;
+ boolean threwEx = true;
try {
store.deleteAll(keys0);
- thewEx = false;
+ threwEx = false;
}
catch (ClassCastException e) {
handleClassCastException(e);
@@ -645,7 +650,7 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
throw new IgniteCheckedException(e);
}
finally {
- sessionEnd0(tx, thewEx);
+ sessionEnd0(tx, threwEx);
}
if (log.isDebugEnabled())
@@ -675,7 +680,12 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
sessionInit0(tx);
try {
- store.sessionEnd(commit);
+ CacheStoreSessionListener sesLsnr = store.getSessionListener();
+
+ if (sesLsnr != null)
+ sesLsnr.onSessionEnd(locSes, commit);
+ else
+ store.sessionEnd(commit);
}
finally {
if (sesHolder != null) {
@@ -715,11 +725,11 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
* @param tx Current transaction.
*/
private void sessionInit0(@Nullable IgniteInternalTx tx) {
- if (sesHolder == null)
- return;
-
+ assert sesHolder != null;
assert sesHolder.get() == null;
+ CacheStoreSessionListener sesLsnr = store.getSessionListener();
+
SessionData ses;
if (tx != null) {
@@ -729,14 +739,21 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
ses = new SessionData(tx, cctx.name());
tx.addMeta(SES_ATTR, ses);
+
+ if (sesLsnr != null)
+ sesLsnr.onSessionStart(locSes);
}
else
// Session cache name may change in cross-cache transaction.
ses.cacheName(cctx.name());
}
- else
+ else {
ses = new SessionData(null, cctx.name());
+ if (sesLsnr != null)
+ sesLsnr.onSessionStart(locSes);
+ }
+
sesHolder.set(ses);
}
@@ -745,8 +762,14 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
*/
private void sessionEnd0(@Nullable IgniteInternalTx tx, boolean threwEx) throws IgniteCheckedException {
try {
- if (tx == null)
- store.sessionEnd(threwEx);
+ if (tx == null) {
+ CacheStoreSessionListener sesLsnr = store.getSessionListener();
+
+ if (sesLsnr != null)
+ sesLsnr.onSessionEnd(locSes, !threwEx);
+ else
+ store.sessionEnd(!threwEx);
+ }
}
catch (Exception e) {
if (!threwEx)
@@ -839,7 +862,14 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
*/
private static class ThreadLocalSession implements CacheStoreSession {
/** */
- private final ThreadLocal<SessionData> sesHolder = new ThreadLocal<>();
+ private final ThreadLocal<SessionData> sesHolder;
+
+ /**
+ * @param sesHolder Session holder.
+ */
+ private ThreadLocalSession(ThreadLocal<SessionData> sesHolder) {
+ this.sesHolder = sesHolder;
+ }
/** {@inheritDoc} */
@Nullable @Override public Transaction transaction() {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
index 5ce42f9..2f1dd89 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheWriteBehindStore.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.store;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.util.tostring.*;
@@ -473,6 +474,11 @@ public class GridCacheWriteBehindStore<K, V> implements CacheStore<K, V>, Lifecy
}
/** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return S.toString(GridCacheWriteBehindStore.class, this);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java
index b14ac5f..8a127f0 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/GridCacheBalancingStoreSelfTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.cache.store;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -197,5 +198,10 @@ public class GridCacheBalancingStoreSelfTest extends GridCommonAbstractTest {
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/cache/store/GridGeneratingTestStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/GridGeneratingTestStore.java b/modules/core/src/test/java/org/apache/ignite/cache/store/GridGeneratingTestStore.java
index 7e9c82e..ba08274 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/GridGeneratingTestStore.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/GridGeneratingTestStore.java
@@ -17,6 +17,7 @@
package org.apache.ignite.cache.store;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.resources.*;
@@ -98,4 +99,9 @@ public class GridGeneratingTestStore implements CacheStore<String, String> {
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
index a3a5717..b7e95fd 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.cache.store;
import org.apache.ignite.*;
import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.lang.*;
@@ -233,5 +234,10 @@ public class IgniteCacheExpiryStoreLoadSelfTest extends GridCacheAbstractSelfTes
@Override public void deleteAll(Collection<?> keys) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
index 157ba93..2af1696 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractLocalStoreSelfTest.java
@@ -21,6 +21,7 @@ import com.google.common.collect.*;
import org.apache.ignite.*;
import org.apache.ignite.cache.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.events.*;
import org.apache.ignite.internal.processors.cache.store.*;
@@ -387,6 +388,11 @@ public abstract class GridCacheAbstractLocalStoreSelfTest extends GridCommonAbst
}
/** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteBiTuple<V, ?> load(K key) throws CacheLoaderException {
return map.get(key);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
index 44171a8..14f8289 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConfigurationConsistencySelfTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.cache.eviction.fifo.*;
import org.apache.ignite.cache.eviction.lru.*;
import org.apache.ignite.cache.eviction.random.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -893,6 +894,11 @@ public class GridCacheConfigurationConsistencySelfTest extends GridCommonAbstrac
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
private static class TestRendezvousAffinityFunction extends RendezvousAffinityFunction {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheGenericTestStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheGenericTestStore.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheGenericTestStore.java
index bc54951..5baced2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheGenericTestStore.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheGenericTestStore.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
@@ -255,6 +256,11 @@ public class GridCacheGenericTestStore<K, V> implements CacheStore<K, V> {
// No-op.
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* Checks the flag and throws exception if it is set. Checks operation delay and sleeps
* for specified amount of time, if needed.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheLifecycleAwareSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheLifecycleAwareSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheLifecycleAwareSelfTest.java
index de5899d..a2b684c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheLifecycleAwareSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheLifecycleAwareSelfTest.java
@@ -21,6 +21,7 @@ import org.apache.ignite.cache.*;
import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.cache.eviction.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.lang.*;
@@ -113,6 +114,11 @@ public class GridCacheLifecycleAwareSelfTest extends GridAbstractLifecycleAwareS
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheStorePutxSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheStorePutxSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheStorePutxSelfTest.java
index a5ed22a..50b515e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheStorePutxSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheStorePutxSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.spi.discovery.tcp.*;
@@ -28,7 +29,6 @@ import org.apache.ignite.testframework.junits.common.*;
import org.apache.ignite.transactions.*;
import org.jetbrains.annotations.*;
-import javax.cache.configuration.*;
import java.util.*;
import java.util.concurrent.atomic.*;
@@ -154,5 +154,10 @@ public class GridCacheStorePutxSelfTest extends GridCommonAbstractTest {
@Override public void sessionEnd(boolean commit) {
// No-op.
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestStore.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestStore.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestStore.java
index 1129e5a..7a60e02 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestStore.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestStore.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.transactions.*;
import org.apache.ignite.internal.util.*;
@@ -303,6 +304,11 @@ public final class GridCacheTestStore implements CacheStore<Integer, String> {
// No-op.
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* Checks the flag and throws exception if it is set. Checks operation delay and sleeps
* for specified amount of time, if needed.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
index 0f2acda..d81800b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
@@ -660,5 +661,10 @@ public abstract class IgniteTxStoreExceptionAbstractSelfTest extends GridCacheAb
if (fail && commit)
throw new CacheWriterException("Store exception");
}
+
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
index cb32b13..bf09f26 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCrossCacheTxStoreSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.distributed;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -290,6 +291,11 @@ public class IgniteCrossCacheTxStoreSelfTest extends GridCommonAbstractTest {
evts.add("deleteAll " + cacheName + " " + keys.size());
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* @return Store session.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
index 7784003..1755a0b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionAbstractTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.integration;
import org.apache.ignite.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -311,6 +312,11 @@ public abstract class IgniteCacheStoreSessionAbstractTest extends IgniteCacheAbs
checkSession("deleteAll");
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* @return Store session.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionWriteBehindAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionWriteBehindAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionWriteBehindAbstractTest.java
index d846881..4e4c74a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionWriteBehindAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/integration/IgniteCacheStoreSessionWriteBehindAbstractTest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache.integration;
import org.apache.ignite.*;
import org.apache.ignite.cache.*;
import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cache.store.tx.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.lang.*;
@@ -204,6 +205,11 @@ public abstract class IgniteCacheStoreSessionWriteBehindAbstractTest extends Ign
checkSession("deleteAll");
}
+ /** {@inheritDoc} */
+ @Override public CacheStoreSessionListener getSessionListener() {
+ return null;
+ }
+
/**
* @return Store session.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/spring/pom.xml
----------------------------------------------------------------------
diff --git a/modules/spring/pom.xml b/modules/spring/pom.xml
index 8494ad0..dbd1f0a 100644
--- a/modules/spring/pom.xml
+++ b/modules/spring/pom.xml
@@ -77,6 +77,12 @@
</dependency>
<dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-jdbc</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
<version>1.1.1</version>
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b3d60836/modules/spring/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSpringSessionListener.java
----------------------------------------------------------------------
diff --git a/modules/spring/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSpringSessionListener.java b/modules/spring/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSpringSessionListener.java
new file mode 100644
index 0000000..d3be25c
--- /dev/null
+++ b/modules/spring/src/main/java/org/apache/ignite/cache/store/tx/CacheStoreSpringSessionListener.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.store.tx;
+
+import org.apache.ignite.cache.store.*;
+import org.springframework.transaction.*;
+import org.springframework.transaction.support.*;
+
+/**
+ * TODO
+ */
+public class CacheStoreSpringSessionListener implements CacheStoreSessionListener {
+ public static final String TX_STATUS_KEY = "__tx_status_";
+
+ private PlatformTransactionManager txMgr;
+
+ public PlatformTransactionManager getTransactionManager() {
+ return txMgr;
+ }
+
+ public void setTransactionManager(PlatformTransactionManager txMgr) {
+ this.txMgr = txMgr;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onSessionStart(CacheStoreSession ses) {
+ assert ses.isWithinTransaction();
+ assert !ses.properties().containsKey(TX_STATUS_KEY);
+
+ DefaultTransactionDefinition def = new DefaultTransactionDefinition();
+
+ def.setIsolationLevel(txIsolation(ses));
+
+ ses.properties().put(TX_STATUS_KEY, txMgr.getTransaction(def));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onSessionEnd(CacheStoreSession ses, boolean commit) {
+ TransactionStatus tx = ses.<String, TransactionStatus>properties().get(TX_STATUS_KEY);
+
+ if (tx != null) {
+ if (commit)
+ txMgr.commit(tx);
+ else
+ txMgr.rollback(tx);
+ }
+ }
+
+ /**
+ * Gets DB transaction isolation level based on ongoing cache transaction isolation.
+ *
+ * @return DB transaction isolation.
+ */
+ private int txIsolation(CacheStoreSession ses) {
+ switch (ses.transaction().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.
+ }
+ }
+}