001package co.codewizards.cloudstore.local;
002
003import static co.codewizards.cloudstore.core.util.AssertUtil.*;
004
005import java.util.HashMap;
006import java.util.Map;
007import java.util.concurrent.CopyOnWriteArrayList;
008import java.util.concurrent.locks.Lock;
009
010import javax.jdo.PersistenceManager;
011import javax.jdo.PersistenceManagerFactory;
012import javax.jdo.Transaction;
013
014import org.slf4j.Logger;
015import org.slf4j.LoggerFactory;
016
017import co.codewizards.cloudstore.core.context.ExtensibleContextSupport;
018import co.codewizards.cloudstore.core.repo.local.ContextWithLocalRepoManager;
019import co.codewizards.cloudstore.core.repo.local.LocalRepoManager;
020import co.codewizards.cloudstore.core.repo.local.LocalRepoTransaction;
021import co.codewizards.cloudstore.core.repo.local.LocalRepoTransactionListenerRegistry;
022import co.codewizards.cloudstore.core.repo.local.LocalRepoTransactionPostCloseEvent;
023import co.codewizards.cloudstore.core.repo.local.LocalRepoTransactionPostCloseListener;
024import co.codewizards.cloudstore.core.repo.local.LocalRepoTransactionPreCloseEvent;
025import co.codewizards.cloudstore.core.repo.local.LocalRepoTransactionPreCloseListener;
026import co.codewizards.cloudstore.core.util.AssertUtil;
027import co.codewizards.cloudstore.local.persistence.Dao;
028import co.codewizards.cloudstore.local.persistence.LocalRepository;
029import co.codewizards.cloudstore.local.persistence.LocalRepositoryDao;
030
031public class LocalRepoTransactionImpl implements LocalRepoTransaction, ContextWithLocalRepoManager, ContextWithPersistenceManager {
032        private static final Logger logger = LoggerFactory.getLogger(LocalRepoTransactionImpl.class);
033
034        private final LocalRepoManager localRepoManager;
035        private final PersistenceManagerFactory persistenceManagerFactory;
036        private final boolean write;
037        private PersistenceManager persistenceManager;
038        private Transaction jdoTransaction;
039        private final Lock lock;
040        private long localRevision = -1;
041        private final Map<Class<?>, Object> daoClass2Dao = new HashMap<>();
042        private final ExtensibleContextSupport extensibleContextSupport = new ExtensibleContextSupport();
043
044        private final LocalRepoTransactionListenerRegistry listenerRegistry = new LocalRepoTransactionListenerRegistry(this);
045
046        private final CopyOnWriteArrayList<LocalRepoTransactionPreCloseListener> preCloseListeners = new CopyOnWriteArrayList<>();
047        private final CopyOnWriteArrayList<LocalRepoTransactionPostCloseListener> postCloseListeners = new CopyOnWriteArrayList<>();
048
049        public LocalRepoTransactionImpl(final LocalRepoManagerImpl localRepoManager, final boolean write) {
050                this.localRepoManager = AssertUtil.assertNotNull(localRepoManager, "localRepoManager");
051                this.persistenceManagerFactory = AssertUtil.assertNotNull(localRepoManager.getPersistenceManagerFactory(), "localRepoManager.persistenceManagerFactory");
052                this.lock = localRepoManager.getLock();
053                this.write = write;
054                begin();
055        }
056
057        private void begin() {
058                lock.lock();
059                try {
060                        if (isActive())
061                                throw new IllegalStateException("Transaction is already active!");
062
063                        lockIfWrite();
064
065                        persistenceManager = persistenceManagerFactory.getPersistenceManager();
066                        jdoTransaction = persistenceManager.currentTransaction();
067                        jdoTransaction.begin();
068                        listenerRegistry.onBegin();
069                } finally {
070                        lock.unlock();
071                }
072        }
073
074        private final void lockIfWrite() {
075                if (write)
076                        lock.lock(); // UNbalance lock to keep it after method returns!
077        }
078
079        private final void unlockIfWrite() {
080                if (write)
081                        lock.unlock(); // UNbalance unlock to counter the unbalanced lock in lockIfWrite().
082        }
083
084        @Override
085        public void commit() {
086                lock.lock();
087                try {
088                        if (!isActive())
089                                throw new IllegalStateException("Transaction is not active!");
090
091                        listenerRegistry.onCommit();
092                        firePreCloseListeners(true);
093                        daoClass2Dao.clear();
094                        jdoTransaction.commit();
095                        persistenceManager.close();
096                        jdoTransaction = null;
097                        persistenceManager = null;
098                        localRevision = -1;
099
100                        unlockIfWrite();
101                } finally {
102                        lock.unlock();
103                }
104                firePostCloseListeners(true);
105        }
106
107        @Override
108        public boolean isActive() {
109                lock.lock();
110                try {
111                        return jdoTransaction != null && jdoTransaction.isActive();
112                } finally {
113                        lock.unlock();
114                }
115        }
116
117        @Override
118        public void rollback() {
119                _rollback();
120                firePostCloseListeners(false);
121        }
122
123        @Override
124        public void rollbackIfActive() {
125                boolean active;
126                lock.lock();
127                try {
128                        active = isActive();
129                        if (active) {
130                                _rollback();
131                        }
132                } finally {
133                        lock.unlock();
134                }
135                if (active) {
136                        firePostCloseListeners(false);
137                }
138        }
139
140        protected void _rollback() {
141                lock.lock();
142                try {
143                        if (!isActive())
144                                throw new IllegalStateException("Transaction is not active!");
145
146                        listenerRegistry.onRollback();
147                        firePreCloseListeners(false);
148                        daoClass2Dao.clear();
149                        jdoTransaction.rollback();
150                        persistenceManager.close();
151                        jdoTransaction = null;
152                        persistenceManager = null;
153                        localRevision = -1;
154
155                        unlockIfWrite();
156                } finally {
157                        lock.unlock();
158                }
159        }
160
161        @Override
162        public void close() {
163                rollbackIfActive();
164        }
165
166        @Override
167        public PersistenceManager getPersistenceManager() {
168                if (!isActive()) {
169                        throw new IllegalStateException("Transaction is not active!");
170                }
171                return persistenceManager;
172        }
173
174        @Override
175        public long getLocalRevision() {
176                if (localRevision < 0) {
177                        if (!write)
178                                throw new IllegalStateException("This is a read-only transaction!");
179
180                        jdoTransaction.setSerializeRead(true);
181                        final LocalRepository lr = getDao(LocalRepositoryDao.class).getLocalRepositoryOrFail();
182                        jdoTransaction.setSerializeRead(null);
183                        localRevision = lr.getRevision() + 1;
184                        lr.setRevision(localRevision);
185                        persistenceManager.flush();
186                }
187                return localRevision;
188        }
189
190        @Override
191        public LocalRepoManager getLocalRepoManager() {
192                return localRepoManager;
193        }
194
195        @Override
196        public <D> D getDao(final Class<D> daoClass) {
197                assertNotNull(daoClass, "daoClass");
198
199                @SuppressWarnings("unchecked")
200                D dao = (D) daoClass2Dao.get(daoClass);
201
202                if (dao == null) {
203                        final PersistenceManager pm = getPersistenceManager();
204                        try {
205                                dao = daoClass.newInstance();
206                        } catch (final InstantiationException e) {
207                                throw new RuntimeException(e);
208                        } catch (final IllegalAccessException e) {
209                                throw new RuntimeException(e);
210                        }
211
212                        if (!(dao instanceof Dao))
213                                throw new IllegalStateException(String.format("dao class %s does not extend Dao!", daoClass.getName()));
214
215                        ((Dao<?, ?>)dao).setPersistenceManager(pm);
216                        ((Dao<?, ?>)dao).setDaoProvider(this);
217
218                        daoClass2Dao.put(daoClass, dao);
219                }
220                return dao;
221        }
222
223        @Override
224        public void flush() {
225                final PersistenceManager pm = getPersistenceManager();
226                pm.flush();
227        }
228
229        @Override
230        public void setContextObject(final Object object) {
231                extensibleContextSupport.setContextObject(object);
232        }
233
234        @Override
235        public <T> T getContextObject(final Class<T> clazz) {
236                return extensibleContextSupport.getContextObject(clazz);
237        }
238
239        @Override
240        public void removeContextObject(Object object) {
241                extensibleContextSupport.removeContextObject(object);
242        }
243
244        @Override
245        public void removeContextObject(Class<?> clazz) {
246                extensibleContextSupport.removeContextObject(clazz);
247        }
248
249        @Override
250        public void addPreCloseListener(LocalRepoTransactionPreCloseListener listener) {
251                preCloseListeners.add(assertNotNull(listener, "listener"));
252        }
253        @Override
254        public void addPostCloseListener(LocalRepoTransactionPostCloseListener listener) {
255                postCloseListeners.add(assertNotNull(listener, "listener"));
256        }
257
258        protected void firePreCloseListeners(final boolean commit) {
259                LocalRepoTransactionPreCloseEvent event = null;
260                for (final LocalRepoTransactionPreCloseListener listener : preCloseListeners) {
261                        try {
262                                if (event == null)
263                                        event = new LocalRepoTransactionPreCloseEvent(this);
264
265                                if (commit)
266                                        listener.preCommit(event);
267                                else
268                                        listener.preRollback(event);
269                        } catch (Exception x) {
270                                logger.error("firePreCloseListeners: " + x, x);
271                        }
272                }
273        }
274        protected void firePostCloseListeners(final boolean commit) {
275                LocalRepoTransactionPostCloseEvent event = null;
276                for (final LocalRepoTransactionPostCloseListener listener : postCloseListeners) {
277                        try {
278                                if (event == null)
279                                        event = new LocalRepoTransactionPostCloseEvent(this);
280
281                                if (commit)
282                                        listener.postCommit(event);
283                                else
284                                        listener.postRollback(event);
285                        } catch (Exception x) {
286                                logger.error("firePostCloseListeners: " + x, x);
287                        }
288                }
289        }
290}