/*
 * Copyright 2010-2012 Luca Garulli (l.garulli--at--orientechnologies.com)
 *
 * Licensed 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 com.orientechnologies.orient.core.storage;

import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicLong;

import com.orientechnologies.common.concur.resource.OCloseable;
import com.orientechnologies.common.concur.resource.OSharedContainerImpl;
import com.orientechnologies.common.concur.resource.OSharedResource;
import com.orientechnologies.common.concur.resource.OSharedResourceAdaptiveExternal;
import com.orientechnologies.common.exception.OException;
import com.orientechnologies.orient.core.Orient;
import com.orientechnologies.orient.core.cache.OLevel2RecordCache;
import com.orientechnologies.orient.core.config.OGlobalConfiguration;
import com.orientechnologies.orient.core.config.OStorageConfiguration;
import com.orientechnologies.orient.core.db.ODatabaseRecordThreadLocal;
import com.orientechnologies.orient.core.exception.OSecurityException;
import com.orientechnologies.orient.core.metadata.schema.OClass;
import com.orientechnologies.orient.core.metadata.security.OSecurityShared;
import com.orientechnologies.orient.core.serialization.serializer.OStringSerializerHelper;

public abstract class OStorageAbstract extends OSharedContainerImpl implements OStorage {
  protected final String                    url;
  protected final String                    mode;
  protected OStorageConfiguration           configuration;
  protected String                          name;
  protected AtomicLong                      version = new AtomicLong();
  protected OLevel2RecordCache              level2Cache;

  protected volatile STATUS                 status  = STATUS.CLOSED;
  protected OSharedResourceAdaptiveExternal lock    = new OSharedResourceAdaptiveExternal(
                                                        OGlobalConfiguration.ENVIRONMENT_CONCURRENT.getValueAsBoolean(), 0, true);

  public OStorageAbstract(final String iName, final String iURL, final String iMode) {
    if (OStringSerializerHelper.contains(iName, '/'))
      name = iName.substring(iName.lastIndexOf("/") + 1);
    else
      name = iName;

    if (OStringSerializerHelper.contains(iName, ','))
      throw new IllegalArgumentException("Invalid character in storage name: " + name);

    level2Cache = new OLevel2RecordCache(this);
    level2Cache.startup();

    url = iURL;
    mode = iMode;
  }

  public OStorageConfiguration getConfiguration() {
    return configuration;
  }

  public boolean isClosed() {
    return status == STATUS.CLOSED;
  }

  public boolean checkForRecordValidity(final OPhysicalPosition ppos) {
    return ppos != null && ppos.recordVersion > -1;
  }

  public String getName() {
    return name;
  }

  /**
   * Returns the configured local Level-2 cache component. Cache component is always created even if not used.
   * 
   * @return
   */
  public OLevel2RecordCache getLevel2Cache() {
    return level2Cache;
  }

  public String getURL() {
    return url;
  }

  public void close() {
    close(false);
  }

  public void close(final boolean iForce) {
    if (!checkForClose(iForce))
      return;

    for (Object resource : sharedResources.values()) {
      if (resource instanceof OSharedResource)
        ((OSharedResource) resource).releaseExclusiveLock();

      if (resource instanceof OCloseable)
        ((OCloseable) resource).close();
    }
    sharedResources.clear();

    Orient.instance().unregisterStorage(this);
  }

  /**
   * Returns current storage's version as serial.
   */
  public long getVersion() {
    return version.get();
  }

  public boolean dropCluster(final String iClusterName) {
    return dropCluster(getClusterIdByName(iClusterName));
  }

  protected boolean checkForClose(final boolean iForce) {
    if (status == STATUS.CLOSED)
      return false;

    final int remainingUsers = getUsers() > 0 ? removeUser() : 0;

    return iForce || (!OGlobalConfiguration.STORAGE_KEEP_OPEN.getValueAsBoolean() && remainingUsers == 0);
  }

  public int getUsers() {
    return lock.getUsers();
  }

  public int addUser() {
    return lock.addUser();
  }

  public int removeUser() {
    return lock.removeUser();
  }

  public OSharedResourceAdaptiveExternal getLock() {
    return lock;
  }

  public long countRecords() {
    long tot = 0;

    for (OCluster c : getClusterInstances())
      if (c != null)
        tot += c.getEntries();

    return tot;
  }

  public <V> V callInLock(final Callable<V> iCallable, final boolean iExclusiveLock) {
    if (iExclusiveLock)
      lock.acquireExclusiveLock();
    else
      lock.acquireSharedLock();
    try {
      return iCallable.call();
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw new OException("Error on nested call in lock", e);
    } finally {
      if (iExclusiveLock)
        lock.releaseExclusiveLock();
      else
        lock.releaseSharedLock();
    }
  }

  @Override
  public String toString() {
    return url != null ? url : "?";
  }

  public STATUS getStatus() {
    return status;
  }

  public void checkForClusterPermissions(final String iClusterName) {
    // CHECK FOR ORESTRICTED
    final Set<OClass> classes = ODatabaseRecordThreadLocal.INSTANCE.get().getMetadata().getSchema()
        .getClassesRelyOnCluster(iClusterName);
    for (OClass c : classes) {
      if (c.isSubClassOf(OSecurityShared.RESTRICTED_CLASSNAME))
        throw new OSecurityException("Class " + c.getName()
            + " cannot be truncated because has record level security enabled (extends " + OSecurityShared.RESTRICTED_CLASSNAME
            + ")");
    }
  }
}

@@ -1567,32 +1567,94 @@
 rityException;%0D%0A
+import com.orientechnologies.orient.core.metadata.OMetadata;%0D%0A
 import com.orien
@@ -6010,33 +6010,26 @@
     
-final Set%3COClass%3E classes
+OMetadata metaData
  = O
@@ -6082,16 +6082,90 @@
 tadata()
+;%0D%0A    if (metaData != null) %7B%0D%0A      final Set%3COClass%3E classes = metaData
 .getSche
@@ -6168,26 +6168,16 @@
 Schema()
-%0D%0A        
 .getClas
@@ -6209,24 +6209,26 @@
 Name);%0D%0A    
+  
 for (OClass 
@@ -6245,24 +6245,26 @@
 s) %7B%0D%0A      
+ 
+ 
 if (c.isSubC
@@ -6314,24 +6314,26 @@
 ))%0D%0A        
+  
 throw new OS
@@ -6373,16 +6373,18 @@
 Name()%0D%0A
+  
         
@@ -6516,16 +6516,18 @@
         
+  
 + %22)%22);%0D
@@ -6523,24 +6523,33 @@
    + %22)%22);%0D%0A
+      %7D%0D%0A
     %7D%0D%0A  %7D%0D%0A

Patch by Sylvain to handle cases without metadata