Main.java

package sample;

import javafx.application.Application;
import javafx.application.Platform;
import javafx.scene.Scene;
import javafx.scene.control.SplitPane;
import javafx.scene.control.TextArea;
import javafx.scene.web.WebEngine;
import javafx.scene.web.WebView;
import javafx.stage.Stage;
import rx.subjects.BehaviorSubject;

import static java.util.concurrent.TimeUnit.*;

public class Main extends Application {

    @Override
    public void start(Stage primaryStage) throws Exception{
        TextArea textArea = new TextArea();
        WebView webView = new WebView();
        WebEngine engine = webView.getEngine();

        BehaviorSubject<String> textSubject = BehaviorSubject.create( "" ) ;
        textArea.textProperty().addListener( ( control, oldValue, newValue ) -> textSubject.onNext( newValue ) ) ;
        textSubject.sample( 500, MILLISECONDS )
                   .distinctUntilChanged()
                   .subscribe( ( s ) -> Platform.runLater( () -> engine.loadContent( s ) ) ) ;
        SplitPane root = new SplitPane();
        root.getItems().addAll(textArea, webView);
        Scene scene = new Scene(root);
        primaryStage.setScene(scene);
        primaryStage.show();
    }


    public static void main(String[] args) {
        launch(args);
    }
}

Yeah, debounce() makes it behave the same and you don't even need distinctUntilChanged(), right?

I believe onCompleted has to originate from the source, not from the subscriber.

Every stream created by reduceCloseSuccessions has it's own Timeline, but I believe all the Timelines use the same scheduler internally.

The distinctUntilChanged would filter out if someone types a word and then deletes the same word (within the debounce period). You don't get an element passing through as the new value is the same as the old one.

Any ideas if this variant would work better for unsubscription ?

Observable<String> textObserver = Observable.<String>create( subscriber -> {
    final ChangeListener<String> textListener = ( value, oldValue, newValue ) -> subscriber.onNext( newValue ) ;
    textArea.textProperty().addListener( textListener ) ;
    return Subscriptions.create( () -> {
        textArea.textProperty().removeListener( textListener ) ;
        System.out.println( "Listener removed" ) ;
    } );
} ).debounce( 500, MILLISECONDS ).distinctUntilChanged() ;
Subscription textAreaSub = textObserver.subscribe( ( s ) -> Platform.runLater( () -> engine.loadContent( s ) ) ;

Then:

textAreaSub.unsubscribe();

As before...

Have you tried this code? Is "Listener removed" printed on textAreaSub.unsubscribe()?

Let me give names to the intermediate Observables for the sake of the text below:

Observable<String> textObserver = Observable.<String>create( subscriber -> {
    final ChangeListener<String> textListener = ( value, oldValue, newValue ) -> subscriber.onNext( newValue ) ;
    textArea.textProperty().addListener( textListener ) ;
    return Subscriptions.create( () -> {
        textArea.textProperty().removeListener( textListener ) ;
        System.out.println( "Listener removed" ) ;
    } );
} );
Observable<String> debounced = textObserver.debounce( 500, MILLISECONDS );
Observable<String> distinct = debounced.distinctUntilChanged() ;
Subscription textAreaSub = distinct.subscribe( ( s ) -> Platform.runLater( () -> engine.loadContent( s ) ) ;

textAreaSub.unsubscribe() clearly unsubscribes from distinct. Does this cause distinct to unsubscribe from debounced, provided that textAreaSub was the only subscriber to distinct? Does that in turn cause debounced to unsubscribe from textObserver, causing the textListener to be removed? I don't know the answer, but this is what ReactFX would do.

I have written a short blogpost about this topic: http://www.guigarage.com/2014/03/reactive-programming-javafx/