gt:             for (int i = 0; i < array.length; i++)
context: /**
   * SMART - State Machine ARchiTecture
   *
   * Copyright (C) 2012 Individual contributors as indicated by
   * the @authors tag
   *
   * This file is a part of SMART.
   *
   * SMART is a free software: you can redistribute it and/or modify
   * it under the terms of the GNU General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * SMART is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
   * GNU General Public License for more details.
   *
   * You should have received a copy of the GNU General Public License
   * along with this program. If not, see <http://www.gnu.org/licenses/>.
   *
   *
   * */
   
  /**
   * ************************************************************
   * HEADERS
   * ************************************************************
   * File:                org.anon.smart.d2cache.store.repository.datasource.pstream.PersistableManager
   * Author:              rsankarx
   * Revision:            1.0
   * Date:                19-07-2016
   *
   * ************************************************************
   * REVISIONS
   * ************************************************************
   * A persistence manager that can be used to read data
   *
   * ************************************************************
   * */
  
  package org.anon.smart.d2cache.store.repository.datasource.pstream;
  
  import java.util.Map;
  import java.util.Set;
  import java.util.HashSet;
  import java.util.Collection;
  import java.util.stream.Stream;
  import java.util.concurrent.ConcurrentHashMap;
  import java.lang.reflect.Type;
  import java.lang.reflect.Field;
  import java.lang.reflect.ParameterizedType;
  
  import org.anon.smart.d2cache.CacheableObject;
  import org.anon.smart.d2cache.store.repository.datasource.DataMetadata;
  import org.anon.smart.d2cache.store.repository.datasource.DataSchema;
  import org.anon.smart.d2cache.store.repository.datasource.DSErrorCodes;
  import org.anon.smart.d2cache.store.repository.datasource.AttributeMetadata;
  import org.anon.smart.d2cache.store.repository.datasource.PersistableData;
  import org.anon.smart.d2cache.store.repository.datasource.pstream.DataReader.QueryData;
  import org.anon.smart.d2cache.store.repository.datasource.pstream.dbutils.DBUtilsReader;
  import org.anon.smart.d2cache.store.repository.datasource.pstream.dbutils.DBUtilsWriter;
  import org.anon.smart.d2cache.store.repository.datasource.resources.ResourceManager.ConnectionPoolEntity;
  
  import org.anon.utilities.exception.CtxException;
  import org.anon.utilities.utils.RepeaterVariants;
  import org.anon.utilities.exception.StandardExceptionHandler;
  
  import static org.anon.utilities.services.ServiceLocator.*;
  
  public class PersistableManager implements AutoCloseable, DSErrorCodes
  {
      static class GlobalThreadLocals
      {
          private static final ThreadLocal<GlobalThreadLocals> LOCALS = new ThreadLocal<GlobalThreadLocals>();
  
          private String _globalName = "";
          private String _dbName = "";
  
          static void setGlobals(String gname, String dbname)
          {
              GlobalThreadLocals gt = LOCALS.get();
              if (gt == null)
              {
                  gt = new GlobalThreadLocals();
              }
              gt._globalName = gname;
              gt._dbName = dbname;
              LOCALS.set(gt);
          }
  
          static String getGlobalName()
          {
              GlobalThreadLocals gt = LOCALS.get();
              if (gt != null)
                  return gt._globalName;
  
              return "";
          }
  
          static String getDBName()
          {
              GlobalThreadLocals gt = LOCALS.get();
              if (gt != null)
                  return gt._dbName;
  
              return "";
          }
      }
  
      public static String globalName() { return GlobalThreadLocals.getGlobalName(); }
      public static String dbName() { return GlobalThreadLocals.getDBName(); }
  
      @FunctionalInterface
      public static interface RepeatParams extends RepeaterVariants
      {
          public Object paramFor(int index);
      }
  
      private static enum dbaccessimpl
      {
          //so we can change this easily if reqd
          dbutils(new DBUtilsReader(), new DBUtilsWriter());
  
          private DataReader _reader;
          private DataWriter _writer;
  
          private dbaccessimpl(DataReader rdr, DataWriter write)
          {
              _reader = rdr;
              _writer = write;
          }
  
          public DataReader getReader(ConnectionPoolEntity cpe)
              throws CtxException
          {
              return (DataReader)_reader.repeatMe(new RepeatParams() {
                  public Object paramFor(int ind) { return cpe; }
              });
          }
  
          public DataWriter getWriter(ConnectionPoolEntity cpe)
              throws CtxException
          {
              return (DataWriter)_writer.repeatMe(new RepeatParams() {
                  public Object paramFor(int ind) { return cpe; }
              });
          }
  
          public DataReader getReader(String dbname)
              throws CtxException
          {
              return (DataReader)_reader.repeatMe(new RepeatParams() { 
                  public Object paramFor(int ind) { return dbname; }
              });
          }
  
          public DataWriter getWriter(String dbname)
              throws CtxException
          {
              return (DataWriter)_writer.repeatMe(new RepeatParams() {
                  public Object paramFor(int ind) { return dbname; }
              });
          }
      }
  
      private static final dbaccessimpl CURRENTIMPL = dbaccessimpl.dbutils;
  
      private static class CacheObjectKey
      {
          private Class _dataclazz;
          private Object _key;
  
          CacheObjectKey(Object obj, Object k)
          {
              _dataclazz = obj.getClass(); 
              _key = k;
          }
  
          CacheObjectKey(Class cls, Object k)
          {
              _dataclazz = cls;
              _key = k;
          }
  
          public int hashCode()
          {
              int hash = _dataclazz.hashCode() + 
                  (31 * _key.hashCode());
              return hash;
          }
  
          public boolean equals(Object obj)
          {
              boolean ret = false;
              if (obj instanceof CacheObjectKey)
              {
                  CacheObjectKey k = (CacheObjectKey)obj;
                  ret = k._dataclazz.equals(_dataclazz);
                  ret = ret && k._key.equals(_key);
              }
  
              return ret;
          }
  
          public String toString()
          {
              return _dataclazz + ":" + _key;
          }
      }
  
      public static class CacheObjectData
      {
          private CacheableObject _object;
          private CacheableObject _original;
          private boolean _isNew;
          private Object[] _keys;
  
          private CacheObjectData(CacheableObject obj, CacheableObject orig, boolean isnew, Object[] keys)
          {
              _object = obj;
              _original = orig;
              _isNew = isnew;
              _keys = keys;
          }
  
          private CacheObjectData(CacheableObject obj, CacheableObject orig, boolean isnew)
          {
              _object = obj;
              _original = orig;
              _isNew = isnew;
          }
  
          public CacheableObject getObject() { return _object; }
          public CacheableObject getOriginal() { return _original; }
          public boolean isNew() { return _isNew; }
          public void resetNew() { _isNew = false; }
          public Object[] getKeys() { return _keys; }
          public Class getObjectClass() 
          { 
              if (_object != null) return _object.getClass();
  
              return null;
          }
      }
  
      private Map<CacheObjectKey, CacheObjectData> _currentObjects;
      private DataReader _streamReader;
      private DataWriter _streamWriter;
      private String _gname;
      private String _dbname;
  
      public PersistableManager(String dbname, String gname) 
          throws CtxException
      {
          this(dbname, CURRENTIMPL, gname);
      }
  
      protected PersistableManager(String dbname, dbaccessimpl impl, String gname)
          throws CtxException
      {
          _currentObjects = new ConcurrentHashMap<CacheObjectKey, CacheObjectData>();
          _streamReader = impl.getReader(dbname);
          _streamWriter = impl.getWriter(dbname);
          _gname = gname;
          _dbname = dbname;
          GlobalThreadLocals.setGlobals(gname, dbname);
      }
  
      public PersistableManager(ConnectionPoolEntity cpe)
          throws CtxException
      {
          _currentObjects = new ConcurrentHashMap<CacheObjectKey, CacheObjectData>();
          _streamReader = CURRENTIMPL.getReader(cpe);
          _streamWriter = CURRENTIMPL.getWriter(cpe);
      }
  
      public <U extends CacheableObject> PersistableManager append(U data, U orig, boolean isnew, Object[] keys)
          throws CtxException
      {
          //Object[] keys = data.smart___keys();
          assertion().assertNotNull(keys, this, INVALID_DATA, "Please provide atleast one key");
          assertion().assertTrue((keys.length > 0), this, INVALID_DATA, "Please provide atleast one key");
          CacheObjectKey put = new CacheObjectKey(data, keys[0]);
  
          CacheObjectData d = new CacheObjectData(data, orig, isnew, keys);
          _currentObjects.put(put, d);
          return this;
      }
  
      private Class getCollectionClass(Field fld)
          throws CtxException
      {
          Class ret = null;
          Type t = fld.getGenericType();
          if (t instanceof ParameterizedType)
          {
              ParameterizedType pt = (ParameterizedType)t;
              int len = pt.getActualTypeArguments().length;
              ret = (Class)pt.getActualTypeArguments()[len - 1];
          }
  
          return ret;
      }
  
      private CacheableObject[] openupData(Object obj, Field fld)
          throws CtxException
      {
          Set<CacheableObject> set = new HashSet<CacheableObject>();
          if (obj instanceof CacheableObject)
          {
              set.add((CacheableObject)obj);
          }
          else if (obj instanceof Collection)
          {
              Collection coll = (Collection)obj;
              assertion().assertNotNull(fld, this, INVALID_DATA, "The data has to be part of an attribute since it is a collection." + obj + ":" + fld);
              Class collcls = getCollectionClass(fld);
              assertion().assertTrue((type().isAssignable(collcls, CacheableObject.class)), this, INVALID_DATA, "The data is not a persistable data." + fld);
              set.addAll(coll);
          }
          else if (obj instanceof Map)
          {
              Map m = (Map)obj;
              assertion().assertNotNull(fld, this, INVALID_DATA, "The data has to be part of an attribute since it is a collection." + obj + ":" + fld);
              Class collcls = getCollectionClass(fld);
              assertion().assertTrue((type().isAssignable(collcls, CacheableObject.class)), this, INVALID_DATA, "The data is not a persistable data." + fld);
              set.addAll(m.values());
          }
          else if (obj.getClass().isArray())
          {
              Object[] array = (Object[])obj;
  