gt:                 for(List<int[]> l : fanOuts) { l.add(next); }
context: /* ---------------------------------------------------------------------
   * Numenta Platform for Intelligent Computing (NuPIC)
   * Copyright (C) 2014, Numenta, Inc.  Unless you have an agreement
   * with Numenta, Inc., for a separate license for this software code, the
   * following terms and conditions apply:
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU General Public License version 3 as
   * published by the Free Software Foundation.
   *
   * This program 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.
   *
   * http://numenta.org/licenses/
   * ---------------------------------------------------------------------
   */
  package org.numenta.nupic.network.sensor;
  
  import gnu.trove.map.TIntObjectMap;
  import gnu.trove.map.hash.TIntObjectHashMap;
  import gnu.trove.map.hash.TObjectIntHashMap;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Spliterator;
  import java.util.Spliterators;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  import java.util.stream.Stream;
  import java.util.stream.StreamSupport;
  
  import org.joda.time.format.DateTimeFormatter;
  import org.numenta.nupic.FieldMetaType;
  import org.numenta.nupic.Parameters;
  import org.numenta.nupic.Parameters.KEY;
  import org.numenta.nupic.algorithms.CLAClassifier;
  import org.numenta.nupic.encoders.AdaptiveScalarEncoder;
  import org.numenta.nupic.encoders.CategoryEncoder;
  import org.numenta.nupic.encoders.CoordinateEncoder;
  import org.numenta.nupic.encoders.DateEncoder;
  import org.numenta.nupic.encoders.DeltaEncoder;
  import org.numenta.nupic.encoders.Encoder;
  import org.numenta.nupic.encoders.Encoder.Builder;
  import org.numenta.nupic.encoders.EncoderTuple;
  import org.numenta.nupic.encoders.GeospatialCoordinateEncoder;
  import org.numenta.nupic.encoders.LogEncoder;
  import org.numenta.nupic.encoders.MultiEncoder;
  import org.numenta.nupic.encoders.RandomDistributedScalarEncoder;
  import org.numenta.nupic.encoders.SDRCategoryEncoder;
  import org.numenta.nupic.encoders.ScalarEncoder;
  import org.numenta.nupic.util.Tuple;
  
  
  /**
   * <p>
   * Decorator for {@link Sensor} types adding HTM
   * specific functionality to sensors and streams.
   * </p><p>
   * The {@code HTMSensor} decorates the sensor with the expected
   * meta data containing field name and field type information 
   * together with the information needed in order to auto-create
   * {@link Encoder}s necessary to output a bit vector specifically
   * tailored for HTM (Hierarchical Temporal Memory) input.
   * </p><p>
   * This class also has very specific date handling capability for
   * the "timestamp" data field type.
   * </p><p>
   * Output is attained by calling {@link #getOutputStream()}. This class
   * extends the Stream API to be able to "fork" streams, so that
   * a single stream can supply multiple fanouts.
   * </p><p>
   * <b>Warning:</b> if {@link #getOutputStream()} is called multiple times,
   * all calls must precede any operations on any of the supplied streams. 
   * </p><p>
   * @author David Ray
   *
   * @param <T>   the input type (i.e. File, URL, etc.)
   */
  public class HTMSensor<T> implements Sensor<T> {
      private Sensor<T> delegate;
      private Header header;
      private Parameters localParameters;
      private MultiEncoder encoder;
      private Stream<int[]> outputStream;
      private List<int[]> output;
      private InputMap inputMap;
      
      private TIntObjectMap<Encoder<?>> indexToEncoderMap;
      private TObjectIntHashMap<String> indexFieldMap = new TObjectIntHashMap<String>();
      
      
      private Iterator<int[]> mainIterator;
      private List<LinkedList<int[]>> fanOuts = new ArrayList<>();
      
      /** Protects {@ #mainIterator} formation and the next() call */
      private Lock criticalAccessLock = new ReentrantLock();
      
      
      
      /**
       * Decorator pattern to construct a new HTMSensor wrapping the specified {@link Sensor}.
       * 
       * @param sensor
       */
      public HTMSensor(Sensor<T> sensor) {
          this.delegate = sensor;
          header = new Header(sensor.getInputStream().getMeta());
          if(header == null || header.size() < 3) {
              throw new IllegalStateException("Header must always be present; and have 3 lines.");
          }
          createEncoder();
      }
      
      /**
       * Called internally during construction to build the encoders
       * needed to process the configured field types.
       */
      @SuppressWarnings("unchecked")
      private void createEncoder() {
          encoder = MultiEncoder.builder().name("MultiEncoder").build();
          
          Map<String, Map<String, Object>> encoderSettings;
          if(localParameters != null && 
              (encoderSettings = (Map<String, Map<String, Object>>)localParameters.getParameterByKey(KEY.FIELD_ENCODING_MAP)) != null &&
                  !encoderSettings.isEmpty()) {
              
              initEncoders(encoderSettings);
              makeIndexEncoderMap();
          }
      }
      
      /**
       * Sets up a mapping which describes the order of occurrence of comma
       * separated fields - mapping their ordinal position to the {@link Encoder}
       * which services the encoding of the field occurring in that position. This
       * sequence of types is contained by an instance of {@link Header} which
       * makes available an array of {@link FieldMetaType}s.
       */
      private void makeIndexEncoderMap() {
          indexToEncoderMap = new TIntObjectHashMap<Encoder<?>>();
          
          for (int i = 0, size = header.getFieldNames().size(); i < size; i++) {
              switch (header.getFieldTypes().get(i)) {
                  case DATETIME:
                      Optional<DateEncoder> de = getDateEncoder(encoder);
                      if (de.isPresent()) {
                          indexToEncoderMap.put(i, de.get());
                      } else {
                          throw new IllegalArgumentException("DateEncoder never initialized: " + header.getFieldNames().get(i));
                      }
                      break;
                  case BOOLEAN:
                  case FLOAT:
                  case INTEGER:
                      Optional<Encoder<?>> ne = getNumberEncoder(encoder);
                      if (ne.isPresent()) {
                          indexToEncoderMap.put(i, ne.get());
                      } else {
                          throw new IllegalArgumentException("Number (or Boolean) encoder never initialized: " + header.getFieldNames().get(i));
                      }
                      break;
                  case LIST:
                  case STRING:
                      Optional<Encoder<?>> ce = getCategoryEncoder(encoder);
                      if (ce.isPresent()) {
                          indexToEncoderMap.put(i, ce.get());
                      } else {
                          throw new IllegalArgumentException("Category encoder never initialized: " + header.getFieldNames().get(i));
                      }
                      break;
                  case COORD:
                  case GEO:
                      Optional<Encoder<?>> ge = getCoordinateEncoder(encoder);
                      if (ge.isPresent()) {
                          indexToEncoderMap.put(i, ge.get());
                      } else {
                          throw new IllegalArgumentException("Coordinate encoder never initialized: " + header.getFieldNames().get(i));
                      }
                      break;
                  default:
                      break;
              }
          }
          
      }
  
      /**
       * Returns an instance of {@link SensorParams} used 
       * to initialize the different types of Sensors with
       * their resource location or source object.
       * 
       * @return a {@link SensorParams} object.
       */
      @Override
      public SensorParams getParams() {
          return delegate.getParams();
      }
  
      /**
       * <p>
       * Main method by which this Sensor's information is retrieved.
       * </p><p>
       * This method returns a subclass of Stream ({@link MetaStream})
       * capable of returning a flag indicating whether a terminal operation
       * has been performed on the stream (i.e. see {@link MetaStream#isTerminal()});
       * in addition the MetaStream returned can return meta information (see
       * {@link MetaStream#getMeta()}.
       * </p>
       * @return  a {@link MetaStream} instance.
       */
      @SuppressWarnings("unchecked")
      @Override
      public <K> MetaStream<K> getInputStream() {
          return (MetaStream<K>)delegate.getInputStream();
      }
      
      /**
       * Customized Iterator which allows "forking" of a Stream
       * into multiple fanouts.
       */
      private class Copy implements Iterator<int[]> {
          private LinkedList<int[]> list;
          Copy(LinkedList<int[]> l) { this.list = l; }
          public boolean hasNext() { return !list.isEmpty() || mainIterator.hasNext(); }
          public int[] next() {
              if(list.isEmpty()) {
                  // We want to make sure only one thread calls next at a time
                  criticalAccessLock.lock();
                  int[] next = mainIterator.next();
  