package no.ssb.vtl.script.support;

import java.util.Comparator;
import java.util.Spliterator;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;

public class JoinSpliterator<L, R, K, O> implements Spliterator<O> {

    final private Comparator<K> comparator;
    final private Spliterator<L> left;
    final private Spliterator<R> right;
    final private Function<L, K> leftKey;
    final private Function<R, K> rightKey;
    final private BiFunction<L, R, O> compute;
    private Pair pair = null;
    private boolean hasNext = false;

    public JoinSpliterator(Comparator<K> comparator, Spliterator<L> left, Spliterator<R> right, Function<L, K> leftKey, Function<R, K> rightKey, BiFunction<L, R, O> compute) {
        this.comparator = comparator;
        this.right = right;
        this.left = left;
        this.leftKey = leftKey;
        this.rightKey = rightKey;
        this.compute = compute;
    }

    boolean advanceLeft() {
        return left.tryAdvance(v -> pair.left = v);
    }

    boolean advanceRight() {
        return right.tryAdvance(v -> pair.right = v);
    }

    @Override
    public boolean tryAdvance(Consumer<? super O> action) {

        if (pair == null) {
            pair = new Pair();
            hasNext = advanceLeft() && advanceRight();
        }

        while (hasNext) {
            int compare = comparator.compare(
                    leftKey.apply(pair.left), rightKey.apply(pair.right)
            );
            if (compare == 0) {
                // generate.
                action.accept(compute.apply(pair.left, pair.right));
                hasNext = advanceLeft() && advanceRight();
                return hasNext;
            } else if (compare < 0) {
                hasNext = advanceLeft();
            } else /* if (compare > 0) */ {
                hasNext = advanceRight();
            }
        }
        return false;
    }

    @Override
    public Spliterator<O> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return Long.max(left.estimateSize(), right.estimateSize());
    }

    @Override
    public int characteristics() {
        return 0;
    }

    private class Pair {
        L left;
        R right;
    }
}

@@ -540,24 +540,31 @@
 ate 
-Pair pair = null
+boolean hadLeft = false
 ;%0A  
@@ -583,20 +583,21 @@
 olean ha
-sNex
+dRigh
 t = fals
@@ -598,16 +598,46 @@
 = false;
+%0A    private Pair pair = null;
 %0A%0A    pu
@@ -1320,36 +1320,36 @@
 ;%0A            ha
-sNex
+dLef
 t = advanceLeft(
@@ -1341,35 +1341,56 @@
  = advanceLeft()
- &&
+;%0A            hadRight =
  advanceRight();
@@ -1418,20 +1418,32 @@
 hile (ha
-sNex
+dLeft %7C%7C hadRigh
 t) %7B%0A   
@@ -1713,36 +1713,36 @@
               ha
-sNex
+dLef
 t = advanceLeft(
@@ -1746,11 +1746,36 @@
 ft()
- &&
+;%0A                hadRight =
  adv
@@ -1810,23 +1810,20 @@
  return 
-hasNext
+true
 ;%0A      
@@ -1864,36 +1864,36 @@
               ha
-sNex
+dLef
 t = advanceLeft(
@@ -1887,32 +1887,65 @@
  advanceLeft();%0A
+                hadRight = true;%0A
             %7D el
@@ -1994,12 +1994,13 @@
   ha
-sNex
+dRigh
 t = 
@@ -2007,32 +2007,64 @@
 advanceRight();%0A
+                hadLeft = true;%0A
             %7D%0A  

Fix JoinSpliterator bug