import org.reactfx.Subscription; //导入依赖的package包/类
public UndoManagerImpl(
        ChangeQueue<C> queue,
        Function<? super C, ? extends C> invert,
        Consumer<C> apply,
        BiFunction<C, C, Optional<C>> merge,
        Predicate<C> isIdentity,
        EventStream<C> changeSource,
        Duration preventMergeDelay) {
    this.queue = queue;
    this.invert = invert;
    this.apply = apply;
    this.merge = merge;
    this.isIdentity = isIdentity;
    this.mark = queue.getCurrentPosition();

    Subscription mainSub = changeSource.subscribe(this::changeObserved);

    if (preventMergeDelay.isZero() || preventMergeDelay.isNegative()) {
        subscription = mainSub;
    } else {
        Subscription sub2 = changeSource.successionEnds(preventMergeDelay).subscribe(ignore -> preventMerge());
        subscription = mainSub.and(sub2);
    }
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
static <T> Val<T> create(
        Supplier<? extends T> computeValue,
        EventStream<?> invalidations) {
    return new ValBase<T>() {

        @Override
        protected Subscription connect() {
            return invalidations.subscribe(x -> invalidate());
        }

        @Override
        protected T computeValue() {
            return computeValue.get();
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Test
public void test() {
    Var<Integer> src = Var.newSimpleVar(1);
    Var<UnaryOperator<Integer>> fn = Var.newSimpleVar(UnaryOperator.identity());
    Val<Integer> mapped = src.mapDynamic(fn);

    assertEquals(1, mapped.getValue().intValue());

    src.setValue(2);
    assertEquals(2, mapped.getValue().intValue());

    fn.setValue(i -> i + i);
    assertEquals(4, mapped.getValue().intValue());

    Subscription sub = mapped.observeChanges((obs, oldVal, newVal) -> {
        assertEquals(4, oldVal.intValue());
        assertEquals(8, newVal.intValue());
    });
    fn.setValue(i -> i * i * i);
    sub.unsubscribe();

    fn.setValue(null);
    assertTrue(mapped.isEmpty());
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Test
public void testLaziness() {
    SimpleVar<String> s1 = (SimpleVar<String>) Var.newSimpleVar("a");
    SimpleVar<String> s2 = (SimpleVar<String>) Var.newSimpleVar("b");
    SimpleVar<String> s3 = (SimpleVar<String>) Var.newSimpleVar("c");

    Val<String> firstNonNull = Val.orElse(s1, s2).orElse(s3);

    assertFalse(s1.isObservingInputs());
    assertFalse(s2.isObservingInputs());
    assertFalse(s2.isObservingInputs());

    Subscription sub = firstNonNull.pin();

    assertTrue(s1.isObservingInputs());
    assertTrue(s2.isObservingInputs());
    assertTrue(s2.isObservingInputs());

    sub.unsubscribe();

    assertFalse(s1.isObservingInputs());
    assertFalse(s2.isObservingInputs());
    assertFalse(s2.isObservingInputs());
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Test
public void test() {
    ObservableList<Integer> list = FXCollections.observableArrayList();
    Val<Integer> size = LiveList.sizeOf(list);
    List<Integer> sizes = new ArrayList<>();
    Subscription sub = EventStreams.valuesOf(size).subscribe(sizes::add);
    list.add(1);
    list.addAll(2, 3, 4);
    assertEquals(Arrays.asList(0, 1, 4), sizes);

    sub.unsubscribe();
    sizes.clear();
    list.addAll(5, 6);
    assertEquals(Arrays.asList(), sizes);

    EventStreams.valuesOf(size).subscribe(sizes::add);
    list.addAll(7, 8);
    assertEquals(Arrays.asList(6, 8), sizes);
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
private static <T> Val<T> avoidFalseInvalidations(Val<T> src) {
    return new ValBase<T>() {
        @Override
        protected T computeValue() {
            return src.getValue();
        }

        @Override
        protected Subscription connect() {
            return src.observeChanges((obs, oldVal,
                                       newVal) -> invalidate());
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
/**
 * It allows to link the content of the file specified once called this method
 */
public void assingFile(String filePath) {

  //action when the button is pressed
  Subscription btnClicksub = EventStreams.eventsOf(this, ActionEvent.ACTION).subscribe(evt -> {
    try {
      LocalList<String> lines = new LocalList<String>();
      lines.addAll(Files.readAllLines(FileSystems.getDefault().getPath(filePath)));
      Utils.clipBoardListener.setContentCB(lines.toString());
    } catch (IOException e) {
      Logger.getGlobal().log(Level.WARNING, e.getMessage());
    }
  });
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
private static <T> Val<T> avoidFalseInvalidations(Val<T> src) {
    return new ValBase<T>() {
        @Override
        protected Subscription connect() {
            return src.observeChanges((obs, oldVal, newVal) -> invalidate());
        }

        @Override
        protected T computeValue() {
            return src.getValue();
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Override
protected Subscription observeInputs() {
    return parChanges.subscribe(mod -> {
        mod = mod.trim();
        QuasiListModification<Paragraph<PS, SEG, S>> qmod =
                QuasiListModification.create(mod.getFrom(), mod.getRemoved(), mod.getAddedSize());
        notifyObservers(qmod.asListChange());
    });
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Override
protected Subscription connect() {
    oldValue = src.getValue();
    return Val.observeChanges(src, (obs, oldVal, newVal) -> {
        oldValue = getValue();
        Duration d = duration.apply(oldValue, newVal);
        transition.setDuration(d);
        transition.playFromStart();
    });
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
/**
 * Returns a stream of invalidated values, which emits the invalidated value
 * (i.e. the old value) on each invalidation of this observable value.
 */
default EventStream<T> invalidations() {
    return new EventStreamBase<T>() {
        @Override
        protected Subscription observeInputs() {
            return observeInvalidations(this::emit);
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
static <T> Subscription observeChanges(
        ObservableValue<? extends T> obs,
        ChangeListener<? super T> listener) {
    if(obs instanceof Val) {
        return ((Val<? extends T>) obs).observeChanges(listener);
    } else {
        obs.addListener(listener);
        return () -> obs.removeListener(listener);
    }
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Override
protected Subscription connect() {
    trySrc = true;
    Subscription sub1 = Val.observeInvalidations(src, obs -> {
        trySrc = true;
        invalidate();
    });
    Subscription sub2 = Val.observeInvalidations(other, obs -> {
        if(!trySrc) {
            invalidate();
        }
    });
    return sub1.and(sub2);
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Override
protected Subscription connect() {
    if(boundTo != null) {
        boundTo.addListener(boundToListener);
    }

    return () -> {
        if(boundTo != null) {
            boundTo.removeListener(boundToListener);
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
@Override
protected Subscription connect() {
    assert tree == null;
    tree = FingerTree.mkTree(input, monoid);
    return LiveList.observeChanges(input, ch -> {
        for(ListModification<? extends T> mod: ch) {
            FingerTree<T, T> left = tree.split(mod.getFrom())._1;
            FingerTree<T, T> right = tree.split(mod.getFrom() + mod.getRemovedSize())._2;
            FingerTree<T, T> middle = FingerTree.mkTree(mod.getAddedSubList(), monoid);
            tree = left.join(middle).join(right);
        }
        invalidate();
    })
    .and(() -> tree = null);
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
default EventStream<QuasiListChange<? extends E>> quasiChanges() {
    return new EventStreamBase<QuasiListChange<? extends E>>() {
        @Override
        protected Subscription observeInputs() {
            return observeQuasiChanges(this::emit);
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
default EventStream<QuasiListModification<? extends E>> quasiModifications() {
    return new EventStreamBase<QuasiListModification<? extends E>>() {
        @Override
        protected Subscription observeInputs() {
            return observeQuasiModifications(this::emit);
        }
    };
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
static <E> Subscription observeQuasiChanges(
        ObservableList<? extends E> list,
        QuasiChangeObserver<? super E> observer) {
    if(list instanceof LiveList) {
        LiveList<? extends E> lst = (LiveList<? extends E>) list;
        return lst.observeQuasiChanges(observer);
    } else {
        ListChangeListener<E> listener = ch -> {
            QuasiListChange<? extends E> change = QuasiListChange.from(ch);
            observer.onChange(change);
        };
        list.addListener(listener);
        return () -> list.removeListener(listener);
    }
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
static <E> Subscription observeChanges(
        ObservableList<E> list,
        Consumer<? super ListChange<? extends E>> observer) {

    return observeQuasiChanges(
            list, qc -> observer.accept(QuasiListChange.instantiate(qc, list)));
}
 

import org.reactfx.Subscription; //导入依赖的package包/类
static <E> EventStream<QuasiListChange<? extends E>> quasiChangesOf(
        ObservableList<E> list) {
    if(list instanceof LiveList) {
        LiveList<E> lst = (LiveList<E>) list;
        return lst.quasiChanges();
    } else {
        return new EventStreamBase<QuasiListChange<? extends E>>() {
            @Override
            protected Subscription observeInputs() {
                return LiveList.<E>observeQuasiChanges(list, this::emit);
            }
        };
    }
}