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 lombok.RequiredArgsConstructor;
  33 import lombok.SneakyThrows;
  34 import lombok.extern.slf4j.Slf4j;
  35 import org.hibernate.HibernateException;
  36 import org.hibernate.LockMode;
  37 import org.hibernate.Session;
  38 import org.hibernate.SessionFactory;
  39 import org.hibernate.cfg.Configuration;
  40 import org.onap.cps.spi.entities.AnchorEntity;
  41 import org.onap.cps.spi.entities.DataspaceEntity;
  42 import org.onap.cps.spi.entities.SchemaSetEntity;
  43 import org.onap.cps.spi.entities.YangResourceEntity;
  44 import org.onap.cps.spi.exceptions.SessionManagerException;
  45 import org.onap.cps.spi.exceptions.SessionTimeoutException;
  46 import org.onap.cps.spi.repository.AnchorRepository;
  47 import org.onap.cps.spi.repository.DataspaceRepository;
  48 import org.springframework.stereotype.Component;
  49
  50 @RequiredArgsConstructor
  51 @Slf4j
  52 @Component
  53 public class SessionManager {
  54
  55     private final TimeLimiterProvider timeLimiterProvider;
  56     private final DataspaceRepository dataspaceRepository;
  57     private final AnchorRepository anchorRepository;
  58     private static SessionFactory sessionFactory;
  59     private static ConcurrentHashMap<String, Session> sessionMap = new ConcurrentHashMap<>();
  60
  61     private synchronized void buildSessionFactory() {
  62         if (sessionFactory == null) {
  63             sessionFactory = new Configuration().configure("hibernate.cfg.xml")
  64                     .addAnnotatedClass(AnchorEntity.class)
  65                     .addAnnotatedClass(DataspaceEntity.class)
  66                     .addAnnotatedClass(SchemaSetEntity.class)
  67                     .addAnnotatedClass(YangResourceEntity.class)
  68                     .buildSessionFactory();
  69         }
  70     }
  71
  72     /**
  73      * Starts a session which allows use of locks and batch interaction with the persistence service.
  74      *
  75      * @return Session ID string
  76      */
  77     public String startSession() {
  78         buildSessionFactory();
  79         final Session session = sessionFactory.openSession();
  80         final String sessionId = UUID.randomUUID().toString();
  81         sessionMap.put(sessionId, session);
  82         session.beginTransaction();
  83         return sessionId;
  84     }
  85
  86     /**
  87      * Close session.
  88      * Locks will be released and changes will be committed.
  89      *
  90      * @param sessionId session ID
  91      */
  92     public void closeSession(final String sessionId) {
  93         try {
  94             final Session session = getSession(sessionId);
  95             session.getTransaction().commit();
  96             session.close();
  97         } catch (final HibernateException e) {
  98             throw new SessionManagerException("Cannot close session",
  99                 String.format("Unable to close session with session ID '%s'", sessionId), e);
 100         } finally {
 101             sessionMap.remove(sessionId);
 102         }
 103     }
 104
 105     /**
 106      * Lock Anchor.
 107      * To release locks(s), the session holding the lock(s) must be closed.
 108      *
 109      * @param sessionId session ID
 110      * @param dataspaceName dataspace name
 111      * @param anchorName anchor name
 112      * @param timeoutInMilliseconds lock attempt timeout in milliseconds
 113      */
 114     @SneakyThrows
 115     public void lockAnchor(final String sessionId, final String dataspaceName,
 116                            final String anchorName, final Long timeoutInMilliseconds) {
 117         final ExecutorService executorService = Executors.newSingleThreadExecutor();
 118         final TimeLimiter timeLimiter = timeLimiterProvider.getTimeLimiter(executorService);
 119
 120         try {
 121             timeLimiter.callWithTimeout(() -> {
 122                 applyPessimisticWriteLockOnAnchor(sessionId, dataspaceName, anchorName);
 123                 return null;
 124             }, timeoutInMilliseconds, TimeUnit.MILLISECONDS);
 125         } catch (final TimeoutException e) {
 126             throw new SessionTimeoutException(
 127                     "Timeout: Anchor locking failed",
 128                     "The error could be caused by another session holding a lock on the specified table. "
 129                             + "Retrying the sending the request could be required.", e);
 130         } catch (final InterruptedException e) {
 131             Thread.currentThread().interrupt();
 132             throw new SessionManagerException("Operation interrupted", "This thread was interrupted.", e);
 133         } catch (final ExecutionException | UncheckedExecutionException e) {
 134             if (e.getCause() != null) {
 135                 throw e.getCause();
 136             }
 137             throw new SessionManagerException(
 138                     "Operation Aborted",
 139                     "The transaction request was aborted. "
 140                             + "Retrying and checking all details are correct could be required", e);
 141         } finally {
 142             executorService.shutdownNow();
 143         }
 144     }
 145
 146     private void applyPessimisticWriteLockOnAnchor(final String sessionId, final String dataspaceName,
 147                                                    final String anchorName) {
 148         final Session session = getSession(sessionId);
 149         final DataspaceEntity dataspaceEntity = dataspaceRepository.getByName(dataspaceName);
 150         final AnchorEntity anchorEntity = anchorRepository.getByDataspaceAndName(dataspaceEntity, anchorName);
 151         final int anchorId = anchorEntity.getId();
 152         log.debug("Attempting to lock anchor {} for session {}", anchorName, sessionId);
 153         session.get(AnchorEntity.class, anchorId, LockMode.PESSIMISTIC_WRITE);
 154         log.info("Anchor {} successfully locked", anchorName);
 155     }
 156
 157     private Session getSession(final String sessionId) {
 158         final Session session = sessionMap.get(sessionId);
 159         if (session == null) {
 160             throw new SessionManagerException("Session not found",
 161                 String.format("Session with ID %s does not exist", sessionId));
 162         }
 163         return session;
 164     }
 165 }