Source code

001/*
002 * (C) Copyright 2006-2016 Nuxeo SA (http://nuxeo.com/) and others.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 *
016 * Contributors:
017 *     George Lefter
018 *     Florent Guillaume
019 */
020package org.nuxeo.ecm.directory.sql;
021
022import java.sql.Connection;
023import java.sql.SQLException;
024import java.util.LinkedHashMap;
025import java.util.LinkedList;
026import java.util.List;
027import java.util.Map;
028import java.util.stream.Collectors;
029
030import javax.transaction.Synchronization;
031
032import org.apache.commons.lang.StringUtils;
033import org.apache.commons.logging.Log;
034import org.apache.commons.logging.LogFactory;
035import org.nuxeo.ecm.core.cache.CacheService;
036import org.nuxeo.ecm.core.schema.SchemaManager;
037import org.nuxeo.ecm.core.schema.types.Field;
038import org.nuxeo.ecm.core.schema.types.Schema;
039import org.nuxeo.ecm.core.storage.sql.ColumnType;
040import org.nuxeo.ecm.core.storage.sql.jdbc.db.Column;
041import org.nuxeo.ecm.core.storage.sql.jdbc.db.Table;
042import org.nuxeo.ecm.core.storage.sql.jdbc.dialect.Dialect;
043import org.nuxeo.ecm.directory.AbstractDirectory;
044import org.nuxeo.ecm.directory.DirectoryCSVLoader;
045import org.nuxeo.ecm.directory.DirectoryException;
046import org.nuxeo.ecm.directory.Session;
047import org.nuxeo.runtime.api.Framework;
048import org.nuxeo.runtime.datasource.ConnectionHelper;
049import org.nuxeo.runtime.transaction.TransactionHelper;
050
051public class SQLDirectory extends AbstractDirectory {
052
053    protected class TxSessionCleaner implements Synchronization {
054        private final SQLSession session;
055
056        Throwable initContext = captureInitContext();
057
058        protected TxSessionCleaner(SQLSession session) {
059            this.session = session;
060        }
061
062        protected Throwable captureInitContext() {
063            if (!log.isDebugEnabled()) {
064                return null;
065            }
066            return new Throwable("SQL directory session init context in " + SQLDirectory.this);
067        }
068
069        protected void checkIsNotLive() {
070            try {
071                if (!session.isLive()) {
072                    return;
073                }
074                if (initContext != null) {
075                    log.warn("Closing a sql directory session for you " + session, initContext);
076                } else {
077                    log.warn("Closing a sql directory session for you " + session);
078                }
079                if (!TransactionHelper.isTransactionActiveOrMarkedRollback()) {
080                    log.warn("Closing sql directory session outside a transaction" + session);
081                }
082                session.close();
083            } catch (DirectoryException e) {
084                log.error("Cannot state on sql directory session before commit " + SQLDirectory.this, e);
085            }
086
087        }
088
089        @Override
090        public void beforeCompletion() {
091            checkIsNotLive();
092        }
093
094        @Override
095        public void afterCompletion(int status) {
096            checkIsNotLive();
097        }
098
099    }
100
101    public static final Log log = LogFactory.getLog(SQLDirectory.class);
102
103    public static final String TENANT_ID_FIELD = "tenantId";
104
105    private final boolean nativeCase;
106
107    private Table table;
108
109    private Schema schema;
110
111    private Map<String, Field> schemaFieldMap;
112
113    // columns to fetch when an entry is read (with the password)
114    protected List<Column> readColumnsAll;
115
116    // columns to fetch when an entry is read (excludes the password)
117    protected List<Column> readColumns;
118
119    // columns to fetch when an entry is read (with the password), as SQL
120    protected String readColumnsAllSQL;
121
122    // columns to fetch when an entry is read (excludes the password), as SQL
123    protected String readColumnsSQL;
124
125    private volatile Dialect dialect;
126
127    public SQLDirectory(SQLDirectoryDescriptor descriptor) {
128        super(descriptor);
129        nativeCase = Boolean.TRUE.equals(descriptor.nativeCase);
130
131        // register the references to other directories
132        addReferences(descriptor.getInverseReferences());
133        addReferences(descriptor.getTableReferences());
134
135        // cache parameterization
136        cache.setEntryCacheName(descriptor.cacheEntryName);
137        cache.setEntryCacheWithoutReferencesName(descriptor.cacheEntryWithoutReferencesName);
138        cache.setNegativeCaching(descriptor.negativeCaching);
139
140        // Cache fallback
141        CacheService cacheService = Framework.getLocalService(CacheService.class);
142        if (cacheService != null) {
143            if (descriptor.cacheEntryName == null && descriptor.getCacheMaxSize() != 0) {
144                cache.setEntryCacheName("cache-" + getName());
145                cacheService.registerCache("cache-" + getName(),
146                        descriptor.getCacheMaxSize(),
147                        descriptor.getCacheTimeout() / 60);
148            }
149            if (descriptor.cacheEntryWithoutReferencesName == null && descriptor.getCacheMaxSize() != 0) {
150                cache.setEntryCacheWithoutReferencesName(
151                        "cacheWithoutReference-" + getName());
152                cacheService.registerCache("cacheWithoutReference-" + getName(),
153                        descriptor.getCacheMaxSize(),
154                        descriptor.getCacheTimeout() / 60);
155            }
156        }
157    }
158
159    @Override
160    public SQLDirectoryDescriptor getDescriptor() {
161        return (SQLDirectoryDescriptor) descriptor;
162    }
163
164    /**
165     * Lazily initializes the connection.
166     *
167     * @return {@code true} if CSV data should be loaded
168     * @since 8.4
169     */
170    protected boolean initConnectionIfNeeded() {
171        // double checked locking with volatile pattern to ensure concurrent lazy init
172        if (dialect == null) {
173            synchronized (this) {
174                if (dialect == null) {
175                    return initConnection();
176                }
177            }
178        }
179        return false;
180    }
181
182    /**
183     * Initializes the table.
184     *
185     * @return {@code true} if CSV data should be loaded
186     * @since 6.0
187     */
188    protected boolean initConnection() {
189        SQLDirectoryDescriptor descriptor = getDescriptor();
190
191        try (Connection sqlConnection = getConnection()) {
192            dialect = Dialect.createDialect(sqlConnection, null);
193            // setup table and fields maps
194            String tableName = descriptor.tableName == null ? descriptor.name : descriptor.tableName;
195            table = SQLHelper.addTable(tableName, dialect, useNativeCase());
196            SchemaManager schemaManager = Framework.getLocalService(SchemaManager.class);
197            schema = schemaManager.getSchema(getSchema());
198            if (schema == null) {
199                throw new DirectoryException("schema not found: " + getSchema());
200            }
201            schemaFieldMap = new LinkedHashMap<>();
202            readColumnsAll = new LinkedList<>();
203            readColumns = new LinkedList<>();
204            boolean hasPrimary = false;
205            for (Field f : schema.getFields()) {
206                String fieldName = f.getName().getLocalName();
207                schemaFieldMap.put(fieldName, f);
208
209                if (!isReference(fieldName)) {
210                    boolean isId = fieldName.equals(getIdField());
211                    ColumnType type = ColumnType.fromField(f);
212                    if (isId && descriptor.isAutoincrementIdField()) {
213                        type = ColumnType.AUTOINC;
214                    }
215                    Column column = SQLHelper.addColumn(table, fieldName, type, useNativeCase());
216                    if (isId) {
217                        if (descriptor.isAutoincrementIdField()) {
218                            column.setIdentity(true);
219                        }
220                        column.setPrimary(true);
221                        column.setNullable(false);
222                        hasPrimary = true;
223                    }
224                    readColumnsAll.add(column);
225                    if (!fieldName.equals(descriptor.passwordField)) {
226                        readColumns.add(column);
227                    }
228                }
229            }
230            readColumnsAllSQL = readColumnsAll.stream().map(Column::getQuotedName).collect(Collectors.joining(", "));
231            readColumnsSQL = readColumns.stream().map(Column::getQuotedName).collect(Collectors.joining(", "));
232            if (!hasPrimary) {
233                throw new DirectoryException(String.format(
234                        "Directory '%s' id field '%s' is not present in schema '%s'", getName(), getIdField(),
235                        getSchema()));
236            }
237
238            SQLHelper helper = new SQLHelper(sqlConnection, table, descriptor.getCreateTablePolicy());
239            boolean loadData = helper.setupTable();
240            return loadData;
241        } catch (SQLException e) {
242            // exception on close
243            throw new DirectoryException(e);
244        }
245    }
246
247    public Connection getConnection() throws DirectoryException {
248        SQLDirectoryDescriptor descriptor = getDescriptor();
249        if (StringUtils.isBlank(descriptor.dataSourceName)) {
250            throw new DirectoryException("Missing dataSource for SQL directory: " + getName());
251        }
252        try {
253            return ConnectionHelper.getConnection(descriptor.dataSourceName);
254        } catch (SQLException e) {
255            throw new DirectoryException("Cannot connect to SQL directory '" + getName() + "': " + e.getMessage(), e);
256        }
257    }
258
259    @Override
260    public Session getSession() throws DirectoryException {
261        boolean loadData = initConnectionIfNeeded();
262        SQLSession session = new SQLSession(this, getDescriptor());
263        addSession(session);
264        if (loadData && descriptor.getDataFileName() != null) {
265            Schema schema = Framework.getService(SchemaManager.class).getSchema(getSchema());
266            DirectoryCSVLoader.loadData(descriptor.getDataFileName(), descriptor.getDataFileCharacterSeparator(),
267                    schema, session::createEntry);
268        }
269        return session;
270    }
271
272    protected void addSession(final SQLSession session) throws DirectoryException {
273        super.addSession(session);
274        registerInTx(session);
275    }
276
277    protected void registerInTx(final SQLSession session) throws DirectoryException {
278        if (!TransactionHelper.isTransactionActive()) {
279            return;
280        }
281        TransactionHelper.registerSynchronization(new TxSessionCleaner(session));
282    }
283
284    public Map<String, Field> getSchemaFieldMap() {
285        return schemaFieldMap;
286    }
287
288    public Table getTable() {
289        return table;
290    }
291
292    public Dialect getDialect() {
293        return dialect;
294    }
295
296    public boolean useNativeCase() {
297        return nativeCase;
298    }
299
300    @Override
301    public boolean isMultiTenant() {
302        return table.getColumn(TENANT_ID_FIELD) != null;
303    }
304
305    @Override
306    public String toString() {
307        return "SQLDirectory [name=" + descriptor.name + "]";
308    }
309
310}