cps-ri/src/main/java/org/onap/cps/spi/utils/SessionManager.java

   1 /*
   2  *  ============LICENSE_START=======================================================
   3  *  Copyright (C) 2022 Nordix Foundation
   4  *  ================================================================================
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *        http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  *
  17  *  SPDX-License-Identifier: Apache-2.0
  18  *  ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.cps.spi.utils;
  22
  23 import com.google.common.util.concurrent.TimeLimiter;
  24 import com.google.common.util.concurrent.UncheckedExecutionException;
  25 import java.util.UUID;
  26 import java.util.concurrent.ConcurrentHashMap;
  27 import java.util.concurrent.ExecutionException;
  28 import java.util.concurrent.ExecutorService;
  29 import java.util.concurrent.Executors;
  30 import java.util.concurrent.TimeUnit;
  31 import java.util.concurrent.TimeoutException;
  32 import javax.annotation.PostConstruct;
  33 import lombok.RequiredArgsConstructor;
  34 import lombok.SneakyThrows;
  35 import lombok.extern.slf4j.Slf4j;
  36 import org.hibernate.HibernateException;
  37 import org.hibernate.LockMode;
  38 import org.hibernate.Session;
  39 import org.onap.cps.spi.config.CpsSessionFactory;
  40 import org.onap.cps.spi.entities.AnchorEntity;
  41 import org.onap.cps.spi.entities.DataspaceEntity;
  42 import org.onap.cps.spi.exceptions.SessionManagerException;
  43 import org.onap.cps.spi.exceptions.SessionTimeoutException;
  44 import org.onap.cps.spi.repository.AnchorRepository;
  45 import org.onap.cps.spi.repository.DataspaceRepository;
  46 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
  47 import org.springframework.context.annotation.Scope;
  48 import org.springframework.stereotype.Component;
  49
  50 @RequiredArgsConstructor
  51 @Slf4j
  52 @Component
  53 @Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
  54 public class SessionManager {
  55
  56     private final CpsSessionFactory cpsSessionFactory;
  57     private final TimeLimiterProvider timeLimiterProvider;
  58     private final DataspaceRepository dataspaceRepository;
  59     private final AnchorRepository anchorRepository;
  60     private final ConcurrentHashMap<String, Session> sessionMap = new ConcurrentHashMap<>();
  61     public static final boolean WITH_COMMIT = true;
  62     public static final boolean WITH_ROLLBACK = false;
  63
  64     @PostConstruct
  65     private void postConstruct() {
  66         final Thread shutdownHook = new Thread(this::closeAllSessionsInShutdown);
  67         Runtime.getRuntime().addShutdownHook(shutdownHook);
  68     }
  69
  70     private void closeAllSessionsInShutdown() {
  71         for (final String sessionId : sessionMap.keySet()) {
  72             try {
  73                 closeSession(sessionId, WITH_ROLLBACK);
  74                 log.info("Session with session ID {} rolled back and closed", sessionId);
  75             } catch (final Exception e) {
  76                 log.warn("Session with session ID {} failed to close", sessionId);
  77             }
  78         }
  79         cpsSessionFactory.closeSessionFactory();
  80     }
  81
  82     /**
  83      * Starts a session which allows use of locks and batch interaction with the persistence service.
  84      *
  85      * @return Session ID string
  86      */
  87     public String startSession() {
  88         final Session session = cpsSessionFactory.openSession();
  89         final String sessionId = UUID.randomUUID().toString();
  90         sessionMap.put(sessionId, session);
  91         session.beginTransaction();
  92         return sessionId;
  93     }
  94
  95     /**
  96      * Close session.
  97      * Changes are committed when commit boolean is set to true.
  98      * Rollback will execute when commit boolean is set to false.
  99      *
 100      * @param sessionId session ID
 101      * @param commit indicator whether session will commit or rollback
 102      */
 103     public void closeSession(final String sessionId, final boolean commit) {
 104         try {
 105             final Session session = getSession(sessionId);
 106             if (commit) {
 107                 session.getTransaction().commit();
 108             } else {
 109                 session.getTransaction().rollback();
 110             }
 111             session.close();
 112         } catch (final HibernateException e) {
 113             throw new SessionManagerException("Cannot close session",
 114                 String.format("Unable to close session with session ID '%s'", sessionId), e);
 115         } finally {
 116             sessionMap.remove(sessionId);
 117         }
 118     }
 119
 120     /**
 121      * Lock Anchor.
 122      * To release locks(s), the session holding the lock(s) must be closed.
 123      *
 124      * @param sessionId session ID
 125      * @param dataspaceName dataspace name
 126      * @param anchorName anchor name
 127      * @param timeoutInMilliseconds lock attempt timeout in milliseconds
 128      */
 129     @SneakyThrows
 130     public void lockAnchor(final String sessionId, final String dataspaceName,
 131                            final String anchorName, final Long timeoutInMilliseconds) {
 132         final ExecutorService executorService = Executors.newSingleThreadExecutor();
 133         final TimeLimiter timeLimiter = timeLimiterProvider.getTimeLimiter(executorService);
 134
 135         try {
 136             timeLimiter.callWithTimeout(() -> {
 137                 applyPessimisticWriteLockOnAnchor(sessionId, dataspaceName, anchorName);
 138                 return null;
 139             }, timeoutInMilliseconds, TimeUnit.MILLISECONDS);
 140         } catch (final TimeoutException e) {
 141             throw new SessionTimeoutException(
 142                     "Timeout: Anchor locking failed",
 143                     "The error could be caused by another session holding a lock on the specified table. "
 144                             + "Retrying the sending the request could be required.", e);
 145         } catch (final InterruptedException e) {
 146             Thread.currentThread().interrupt();
 147             throw new SessionManagerException("Operation interrupted", "This thread was interrupted.", e);
 148         } catch (final ExecutionException | UncheckedExecutionException e) {
 149             if (e.getCause() != null) {
 150                 throw e.getCause();
 151             }
 152             throw new SessionManagerException(
 153                     "Operation Aborted",
 154                     "The transaction request was aborted. "
 155                             + "Retrying and checking all details are correct could be required", e);
 156         } finally {
 157             executorService.shutdownNow();
 158         }
 159     }
 160
 161     private void applyPessimisticWriteLockOnAnchor(final String sessionId, final String dataspaceName,
 162                                                    final String anchorName) {
 163         final Session session = getSession(sessionId);
 164         final DataspaceEntity dataspaceEntity = dataspaceRepository.getByName(dataspaceName);
 165         final AnchorEntity anchorEntity = anchorRepository.getByDataspaceAndName(dataspaceEntity, anchorName);
 166         final int anchorId = anchorEntity.getId();
 167         log.debug("Attempting to lock anchor {} for session {}", anchorName, sessionId);
 168         session.get(AnchorEntity.class, anchorId, LockMode.PESSIMISTIC_WRITE);
 169         log.info("Anchor {} successfully locked", anchorName);
 170     }
 171
 172     private Session getSession(final String sessionId) {
 173         final Session session = sessionMap.get(sessionId);
 174         if (session == null) {
 175             throw new SessionManagerException("Session not found",
 176                 String.format("Session with ID %s does not exist", sessionId));
 177         }
 178         return session;
 179     }
 180
 181 }