gt:             if (type == null) {
context: /**************************************************************************************
   * Copyright (C) 2006-2015 EsperTech Inc. All rights reserved.                        *
   * http://www.espertech.com/esper                                                          *
   * http://www.espertech.com                                                           *
   * ---------------------------------------------------------------------------------- *
   * The software in this package is published under the terms of the GPL license       *
   * a copy of which has been included with this distribution in the license.txt file.  *
   **************************************************************************************/
  package com.espertech.esper.epl.variable;
  
  import com.espertech.esper.client.EventBean;
  import com.espertech.esper.client.EventType;
  import com.espertech.esper.client.VariableValueException;
  import com.espertech.esper.collection.Pair;
  import com.espertech.esper.core.service.StatementExtensionSvcContext;
  import com.espertech.esper.epl.core.EngineImportException;
  import com.espertech.esper.epl.core.EngineImportService;
  import com.espertech.esper.event.EventAdapterService;
  import com.espertech.esper.schedule.TimeProvider;
  import com.espertech.esper.util.JavaClassHelper;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import java.io.StringWriter;
  import java.util.*;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.CopyOnWriteArraySet;
  import java.util.concurrent.locks.ReadWriteLock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  /**
   * Variables service for reading and writing variables, and for setting a version number for the current thread to
   * consider variables for.
   * <p>
   * Consider a statement as follows: select * from MyEvent as A where A.val > var1 and A.val2 > var1 and A.val3 > var2
   * <p>
   * Upon statement execution we need to guarantee that the same atomic value for all variables is applied for all
   * variable reads (by expressions typically) within the statement.
   * <p>
   * Designed to support:
   * <ol>
   * <li>lock-less read of the current and prior version, locked reads for older versions
   * <li>atomicity by keeping multiple versions for each variable and a threadlocal that receives the current version each call
   * <li>one write lock for all variables (required to coordinate with single global version number),
   *   however writes are very fast (entry to collection plus increment an int) and therefore blocking should not be an issue
   * </ol>
   * <p>
   * As an alternative to a version-based design, a read-lock for the variable space could also be used, with the following
   * disadvantages: The write lock may just not be granted unless fair locks are used which are more expensive; And
   * a read-lock is more expensive to acquire for multiple CPUs; A thread-local is still need to deal with
   * "set var1=3, var2=var1+1" assignments where the new uncommitted value must be visible in the local evaluation.
   * <p>
   * Every new write to a variable creates a new version. Thus when reading variables, readers can ignore newer versions
   * and a read lock is not required in most circumstances.
   * <p>
   * This algorithm works as follows:
   * <p>
   * A thread processing an event into the engine via sendEvent() calls the "setLocalVersion" method once
   * before processing a statement that has variables.
   * This places into a threadlocal variable the current version number, say version 570.
   * <p>
   * A statement that reads a variable has an {@link com.espertech.esper.epl.expression.core.ExprVariableNode} that has a {@link com.espertech.esper.epl.variable.VariableReader} handle
   * obtained during validation (example).
   * <p>
   * The {@link com.espertech.esper.epl.variable.VariableReader} takes the version from the threadlocal (570) and compares the version number with the
   * version numbers held for the variable.
   * If the current version is same or lower (520, as old or older) then the threadlocal version,
   * then use the current value.
   * If the current version is higher (571, newer) then the threadlocal version, then go to the prior value.
   * Use the prior value until a version is found that as old or older then the threadlocal version.
   * <p>
   * If no version can be found that is old enough, output a warning and return the newest version.
   * This should not happen, unless a thread is executing for very long within a single statement such that
   * lifetime-old-version time speriod passed before the thread asks for variable values.
   * <p>
   * As version numbers are counted up they may reach a boundary. Any write transaction after the boundary
   * is reached performs a roll-over. In a roll-over, all variables version lists are
   * newly created and any existing threads that read versions go against a (old) high-collection,
   * while new threads reading the reset version go against a new low-collection.
   * <p>
   * The class also allows an optional state handler to be plugged in to handle persistence for variable state.
   * The state handler gets invoked when a variable changes value, and when a variable gets created
   * to obtain the current value from persistence, if any.
   */
  public class VariableServiceImpl implements VariableService
  {
      private static Log log = LogFactory.getLog(VariableServiceImpl.class);
  
      /**
       * Sets the boundary above which a reader considers the high-version list of variable values.
       * For use in roll-over when the current version number overflows the ROLLOVER_WRITER_BOUNDARY.
       */
      protected final static int ROLLOVER_READER_BOUNDARY = Integer.MAX_VALUE - 100000;
  
      /**
       * Applicable for each variable if more then the number of versions accumulated, check
       * timestamps to determine if a version can be expired.
       */
      protected final static int HIGH_WATERMARK_VERSIONS = 50;
  
      // Each variable has an index number, a context-partition id, a current version and a list of values
      private final ArrayList<ConcurrentHashMap<Integer, VariableReader>> variableVersionsPerCP;
  
      // Each variable and a context-partition id may have a set of callbacks to invoke when the variable changes
      private final ArrayList<Map<Integer, Set<VariableChangeCallback>>> changeCallbacksPerCP;
  
      // Keep the variable list
      private final Map<String, VariableMetaData> variables;
  
      // Write lock taken on write of any variable; and on read of older versions
      private final ReadWriteLock readWriteLock;
  
      // Thread-local for the visible version per thread
      private VariableVersionThreadLocal versionThreadLocal = new VariableVersionThreadLocal();
  
      // Number of milliseconds that old versions of a variable are allowed to live
      private final long millisecondLifetimeOldVersions;
      private final TimeProvider timeProvider;
      private final EventAdapterService eventAdapterService;
      private final VariableStateHandler optionalStateHandler;
  
      private volatile int currentVersionNumber;
      private int currentVariableNumber;
  
      /**
       * Ctor.
       * @param millisecondLifetimeOldVersions number of milliseconds a version may hang around before expiry
       * @param timeProvider provides the current time
       * @param optionalStateHandler a optional plug-in that may store variable state and retrieve state upon creation
       * @param eventAdapterService event adapters
       */
      public VariableServiceImpl(long millisecondLifetimeOldVersions, TimeProvider timeProvider, EventAdapterService eventAdapterService, VariableStateHandler optionalStateHandler)
      {
          this(0, millisecondLifetimeOldVersions, timeProvider, eventAdapterService, optionalStateHandler);
      }
  
      /**
       * Ctor.
       * @param startVersion the first version number to start from
       * @param millisecondLifetimeOldVersions number of milliseconds a version may hang around before expiry
       * @param timeProvider provides the current time
       * @param optionalStateHandler a optional plug-in that may store variable state and retrieve state upon creation
       * @param eventAdapterService for finding event types
       */
      protected VariableServiceImpl(int startVersion, long millisecondLifetimeOldVersions, TimeProvider timeProvider, EventAdapterService eventAdapterService, VariableStateHandler optionalStateHandler)
      {
          this.millisecondLifetimeOldVersions = millisecondLifetimeOldVersions;
          this.timeProvider = timeProvider;
          this.eventAdapterService = eventAdapterService;
          this.optionalStateHandler = optionalStateHandler;
          this.variables = new HashMap<String, VariableMetaData>();
          this.readWriteLock = new ReentrantReadWriteLock();
          this.variableVersionsPerCP = new ArrayList<ConcurrentHashMap<Integer, VariableReader>>();
          this.changeCallbacksPerCP = new ArrayList<Map<Integer, Set<VariableChangeCallback>>>();
          currentVersionNumber = startVersion;
      }
  
      public void destroy() {
          versionThreadLocal = new VariableVersionThreadLocal();
      }
  
      public synchronized void removeVariableIfFound(String name) {
          VariableMetaData metaData = variables.get(name);
          if (metaData == null) {
              return;
          }
  
          if (log.isDebugEnabled()) {
              log.debug("Removing variable '" + name + "'");
          }
          variables.remove(name);
  
          if (optionalStateHandler != null) {
              ConcurrentHashMap<Integer, VariableReader> readers = variableVersionsPerCP.get(metaData.getVariableNumber());
              Set<Integer> cps = Collections.emptySet();
              if (readers != null) {
                  cps = readers.keySet();
              }
              optionalStateHandler.removeVariable(name, cps);
          }
  
          int number = metaData.getVariableNumber();
          variableVersionsPerCP.set(number, null);
          changeCallbacksPerCP.set(number, null);
      }
  
      public void setLocalVersion()
      {
          versionThreadLocal.getCurrentThread().setVersion(currentVersionNumber);
      }
  
      public void registerCallback(String variableName, int agentInstanceId, VariableChangeCallback variableChangeCallback)
      {
          VariableMetaData metaData = variables.get(variableName);
          if (metaData == null) {
              return;
          }
  
          Map<Integer, Set<VariableChangeCallback>> cps = changeCallbacksPerCP.get(metaData.getVariableNumber());
          if (cps == null) {
              cps = new HashMap<Integer, Set<VariableChangeCallback>>();
              changeCallbacksPerCP.set(metaData.getVariableNumber(), cps);
          }
  
          if (metaData.getContextPartitionName() == null) {
              agentInstanceId = 0;
          }
  
          Set<VariableChangeCallback> callbacks = cps.get(agentInstanceId);
          if (callbacks == null) {
              callbacks = new CopyOnWriteArraySet<VariableChangeCallback>();
              cps.put(agentInstanceId, callbacks);
          }
          callbacks.add(variableChangeCallback);
      }
  
      public void unregisterCallback(String variableName, int agentInstanceId, VariableChangeCallback variableChangeCallback)
      {
          VariableMetaData metaData = variables.get(variableName);
          if (metaData == null) {
              return;
          }
  
          Map<Integer, Set<VariableChangeCallback>> cps = changeCallbacksPerCP.get(metaData.getVariableNumber());
          if (cps == null) {
              return;
          }
  
          if (metaData.getContextPartitionName() == null) {
              agentInstanceId = 0;
          }
  
          Set<VariableChangeCallback> callbacks = cps.get(agentInstanceId);
          if (callbacks != null) {
              callbacks.remove(variableChangeCallback);
          }
      }
  
      public void createNewVariable(String optionalContextName, String variableName, String variableType, boolean constant, boolean array, boolean arrayOfPrimitive, Object value, EngineImportService engineImportService) throws VariableExistsException, VariableTypeException
      {
          // Determime the variable type
          Class primitiveType = JavaClassHelper.getPrimitiveClassForName(variableType);
          Class type = JavaClassHelper.getClassForSimpleName(variableType);
          Class arrayType = null;
          EventType eventType = null;
          if (type == null) {
              if (variableType.toLowerCase().equals("object")) {
                  type = Object.class;
              }
  