public OfLong iterator() {
  DurableInput in = pool.instance();

  return new OfLong() {
    boolean hasNext = true;
    int next = (int) in.readVLQ();

    @Override
    public long nextLong() {
      long result = next;
      if (in.remaining() > 0) {
        next += (int) in.readUVLQ();
      } else {
        hasNext = false;
      }
      return result;
    }

    @Override
    public boolean hasNext() {
      return hasNext;
    }
  };
}
 

public IndexRange candidateIndices(long hash) {
  int start = -1, end = -1;
  OfLong it = iterator();

  for (int i = 0; it.hasNext(); i++) {
    long curr = it.nextLong();
    if (curr == hash) {
      start = i;
      break;
    } else if (curr > hash) {
      return new IndexRange(-1, -1, true);
    }
  }

  for (end = start; it.hasNext(); end++) {
    if (it.nextLong() > hash) {
      return new IndexRange(start, end + 1, true);
    }
  }

  return new IndexRange(start, end + 1, false);
}
 

/**
 * Test write long read long unsigned
 *
 */
@Test
public void ReadWriteLongTestUnsigned() {
    ByteBuffer bb = ByteBuffer.allocate(128);
    Random r= new Random();
    r.setSeed(0);
    OfLong randomStream = r.longs(0L, Long.MAX_VALUE).iterator();
    for (int i = 1; i < Long.BYTES; i++) {
        for (int l = 0; l < LOOP_COUNT; l++) {
            long value = randomStream.nextLong() >> (i * Byte.SIZE);
            HTVarInt.writeLong(bb, value);
            bb.position(0);
            assertEquals(l + " " + Long.toHexString(value), value, HTVarInt.readLong(bb));
            bb.position(0);
        }
    }
}
 

@Override
  public OfLong iterator() {
    return new OfLong() {
      int i = 0;
      
      @Override
      public boolean hasNext() {
        return i < size();
      }

//      @Override   // using default
//      public Long next() {
//        if (!hasNext())
//          throw new NoSuchElementException();
//        return get(i++);
//      }

      @Override
      public long nextLong() {
        if (!hasNext())
          throw new NoSuchElementException();
        return get(i++);
      }
   };
  }
 

@Test
public void testDropWhile() {
    assertArrayEquals(new long[] { 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 }, LongStreamEx.range(100).dropWhile(
        i -> i % 10 < 5).limit(10).toArray());
    assertEquals(100, LongStreamEx.range(100).dropWhile(i -> i % 10 < 0).count());
    assertEquals(0, LongStreamEx.range(100).dropWhile(i -> i % 10 < 10).count());
    assertEquals(OptionalLong.of(0), LongStreamEx.range(100).dropWhile(i -> i % 10 < 0).findFirst());
    assertEquals(OptionalLong.empty(), LongStreamEx.range(100).dropWhile(i -> i % 10 < 10).findFirst());

    java.util.Spliterator.OfLong spltr = LongStreamEx.range(100).dropWhile(i -> i % 10 < 1).spliterator();
    assertTrue(spltr.tryAdvance((long x) -> assertEquals(1, x)));
    Builder builder = LongStream.builder();
    spltr.forEachRemaining(builder);
    assertArrayEquals(LongStreamEx.range(2, 100).toArray(), builder.build().toArray());
}
 

public long nth(long index) {
  OfLong it = iterator();
  Iterators.drop(it, index);
  return it.nextLong();
}
 

@Override
public void run(String... args) throws Exception {
	
	getCloudEnvProperties();
	loadPoSCounties();
	
	if (!skipSetup){
		runSetup();
	}
	if (numberOfTransactions<0) numberOfTransactions = Integer.MAX_VALUE;
	
	logger.info(">>>>> RUNNING SIMULATION");		
	logger.info("--------------------------------------");
	logger.info(">>> Geode rest endpoint: "+geodeURL);
	logger.info("--------------------------------------");
	
	logger.info(">>> Posting "+numberOfTransactions+" transactions ...");

	int numberOfDevices = counties.size();
	
	OfLong deviceIDs = new Random().longs(0, numberOfDevices).iterator();
	OfLong accountIDs = new Random().longs(0, numberOfAccounts).iterator();
	
	Random random = new Random();
	long mean = 100; // mean value for transactions
	long variance = 40; // variance

	DecimalFormat df = new DecimalFormat();
	df.setMaximumFractionDigits(2);
	
	for (int i=0; i<numberOfTransactions; i++){
		//Map<String,Object> map = (Map)objects.get(i);
		Transaction t = new Transaction();
		t.setId(Math.abs(UUID.randomUUID().getLeastSignificantBits()));
		long accountId = accountIDs.next();
		t.setAccountId(accountId);		
		
		// 90% of times, we'll transact this account from a single "home location"
		if (Math.random()<0.9){
			t.setDeviceId(getHomePoS(accountId));
		}
		else {
			t.setDeviceId(deviceIDs.next());
		}
		
		
		
		t.setTimestamp(System.currentTimeMillis());
		
		double value = Double.parseDouble(df.format(Math.abs(mean+random.nextGaussian()*variance)));  
		t.setValue(value);
		
		try{					
			Transaction response = restTemplate.postForObject(geodeURL+RegionName.Transaction, t, Transaction.class);
		}
		catch(Exception e){
			logger.warning("Failed to connect to Geode using URL "+geodeURL);
			e.printStackTrace();
		}
		Thread.sleep(delay);
		

	}

	logger.info("done");
	
	
}
 

LongStreamEx(Spliterator.OfLong spliterator, StreamContext context) {
    super(spliterator, context);
}
 

final LongStreamEx delegate(Spliterator.OfLong spliterator) {
    return new LongStreamEx(spliterator, context);
}
 

@Override
public OfLong iterator() {
    return Spliterators.iterator(spliterator());
}
 

@Test
public void testBasics() {
    assertFalse(LongStreamEx.of(1).isParallel());
    assertTrue(LongStreamEx.of(1).parallel().isParallel());
    assertFalse(LongStreamEx.of(1).parallel().sequential().isParallel());
    AtomicInteger i = new AtomicInteger();
    try (LongStreamEx s = LongStreamEx.of(1).onClose(i::incrementAndGet)) {
        assertEquals(1, s.count());
    }
    assertEquals(1, i.get());
    assertEquals(6, LongStreamEx.range(0, 4).sum());
    assertEquals(3, LongStreamEx.range(0, 4).max().getAsLong());
    assertEquals(0, LongStreamEx.range(0, 4).min().getAsLong());
    assertEquals(1.5, LongStreamEx.range(0, 4).average().getAsDouble(), 0.000001);
    assertEquals(4, LongStreamEx.range(0, 4).summaryStatistics().getCount());
    assertArrayEquals(new long[] { 1, 2, 3 }, LongStreamEx.range(0, 5).skip(1).limit(3).toArray());
    assertArrayEquals(new long[] { 1, 2, 3 }, LongStreamEx.of(3, 1, 2).sorted().toArray());
    assertArrayEquals(new long[] { 1, 2, 3 }, LongStreamEx.of(1, 2, 1, 3, 2).distinct().toArray());
    assertArrayEquals(new int[] { 2, 4, 6 }, LongStreamEx.range(1, 4).mapToInt(x -> (int) x * 2).toArray());
    assertArrayEquals(new long[] { 2, 4, 6 }, LongStreamEx.range(1, 4).map(x -> x * 2).toArray());
    assertArrayEquals(new double[] { 2, 4, 6 }, LongStreamEx.range(1, 4).mapToDouble(x -> x * 2).toArray(), 0.0);
    assertArrayEquals(new long[] { 1, 3 }, LongStreamEx.range(0, 5).filter(x -> x % 2 == 1).toArray());
    assertEquals(6, LongStreamEx.of(1, 2, 3).reduce(Long::sum).getAsLong());
    assertEquals(Long.MAX_VALUE, LongStreamEx.rangeClosed(1, Long.MAX_VALUE).spliterator().getExactSizeIfKnown());

    assertTrue(LongStreamEx.of(1, 2, 3).spliterator().hasCharacteristics(Spliterator.ORDERED));
    assertFalse(LongStreamEx.of(1, 2, 3).unordered().spliterator().hasCharacteristics(Spliterator.ORDERED));

    OfLong iterator = LongStreamEx.of(1, 2, 3).iterator();
    assertEquals(1L, iterator.nextLong());
    assertEquals(2L, iterator.nextLong());
    assertEquals(3L, iterator.nextLong());
    assertFalse(iterator.hasNext());

    AtomicInteger idx = new AtomicInteger();
    long[] result = new long[500];
    LongStreamEx.range(1000).atLeast(500).parallel().forEachOrdered(val -> result[idx.getAndIncrement()] = val);
    assertArrayEquals(LongStreamEx.range(500, 1000).toArray(), result);

    assertTrue(LongStreamEx.empty().noneMatch(x -> true));
    assertFalse(LongStreamEx.of(1).noneMatch(x -> true));
    assertTrue(LongStreamEx.of(1).noneMatch(x -> false));
}
 

@Override
public Spliterator.OfLong spliterator() {
  return Arrays.spliterator(theArray);
}
 

/**
 * Produces an array containing cumulative results of applying the
 * accumulation function going left to right.
 * 
 * <p>
 * This is a terminal operation.
 * 
 * <p>
 * For parallel stream it's not guaranteed that accumulator will always be
 * executed in the same thread.
 * 
 * <p>
 * This method cannot take all the advantages of parallel streams as it must
 * process elements strictly left to right.
 *
 * @param accumulator a
 *        <a href="package-summary.html#NonInterference">non-interfering
 *        </a>, <a href="package-summary.html#Statelessness">stateless</a>
 *        function for incorporating an additional element into a result
 * @return the array where the first element is the first element of this
 *         stream and every successor element is the result of applying
 *         accumulator function to the previous array element and the
 *         corresponding stream element. The resulting array has the same
 *         length as this stream.
 * @see #foldLeft(LongBinaryOperator)
 * @since 0.5.1
 */
public long[] scanLeft(LongBinaryOperator accumulator) {
    Spliterator.OfLong spliterator = spliterator();
    int size = intSize(spliterator);
    LongBuffer buf = new LongBuffer(size >= 0 ? size : INITIAL_SIZE);
    delegate(spliterator).forEachOrdered(i -> buf.add(buf.size == 0 ? i
            : accumulator.applyAsLong(buf.data[buf.size - 1], i)));
    return buf.toArray();
}
 

/**
 * Returns a stream containing cumulative results of applying the
 * accumulation function going left to right.
 * 
 * <p>
 * This is a stateful
 * <a href="package-summary.html#StreamOps">quasi-intermediate</a>
 * operation.
 *
 * <p>
 * This operation resembles {@link #scanLeft(LongBinaryOperator)}, but
 * unlike {@code scanLeft} this operation is intermediate and accumulation
 * function must be associative.
 * 
 * <p>
 * This method cannot take all the advantages of parallel streams as it must
 * process elements strictly left to right. Using an unordered source or
 * removing the ordering constraint with {@link #unordered()} may improve
 * the parallel processing speed.
 *
 * @param op an <a href="package-summary.html#Associativity">associative</a>
 *        , <a href="package-summary.html#NonInterference">non-interfering
 *        </a>, <a href="package-summary.html#Statelessness">stateless</a>
 *        function for computing the next element based on the previous one
 * @return the new stream.
 * @see #scanLeft(LongBinaryOperator)
 * @since 0.6.1
 */
public LongStreamEx prefix(LongBinaryOperator op) {
    return delegate(new PrefixOps.OfLong(spliterator(), op));
}
 

/**
 * Returns a sequential {@code LongStreamEx} containing a single element.
 *
 * @param element the single element
 * @return a singleton sequential stream
 */
public static LongStreamEx of(long element) {
    return of(new ConstSpliterator.OfLong(element, 1, true));
}
 

/**
 * Returns a sequential {@link LongStreamEx} created from given
 * {@link java.util.Spliterator.OfLong}.
 * 
 * @param spliterator a spliterator to create the stream from.
 * @return the new stream
 * @since 0.3.4
 */
public static LongStreamEx of(Spliterator.OfLong spliterator) {
    return new LongStreamEx(spliterator, StreamContext.SEQUENTIAL);
}
 

/**
 * Returns a sequential, ordered {@link LongStreamEx} created from given
 * {@link java.util.PrimitiveIterator.OfLong}.
 *
 * <p>
 * This method is roughly equivalent to
 * {@code LongStreamEx.of(Spliterators.spliteratorUnknownSize(iterator, ORDERED))}
 * , but may show better performance for parallel processing.
 * 
 * <p>
 * Use this method only if you cannot provide better Stream source.
 *
 * @param iterator an iterator to create the stream from.
 * @return the new stream
 * @since 0.5.1
 */
public static LongStreamEx of(PrimitiveIterator.OfLong iterator) {
    return of(new UnknownSizeSpliterator.USOfLong(iterator));
}
 

/**
 * Returns a sequential unordered {@code LongStreamEx} of given length which
 * elements are equal to supplied value.
 * 
 * @param value the constant value
 * @param length the length of the stream
 * @return a new {@code LongStreamEx}
 * @since 0.1.2
 */
public static LongStreamEx constant(long value, long length) {
    return of(new ConstSpliterator.OfLong(value, length, false));
}
 

/**
 * Returns the spliterator which covers all the elements emitted by this
 * emitter.
 * 
 * @return the new spliterator
 */
default Spliterator.OfLong spliterator() {
    return new EmitterSpliterator.OfLong(this);
}