/**
 * Slowly request numbers
 */
@FXML
public void startBackpressure(ActionEvent actionEvent) {
    resetGraph();
    Single.just(5000)
            // Construct request
            .map(i -> NumbersProto.HowMany.newBuilder().setNumber(i).build())
            // Call service
            .as(stub::sendNumbers)
            // Parse response
            .map(i -> i.getNumber(0))
            // Introduce a synthetic three millisecond delay per read
            .zipWith(Flowable.interval(3, TimeUnit.MILLISECONDS), (item, interval) -> item)
            // Transition processing to UI thread
            .observeOn(JavaFxScheduler.platform())
            // Execute
            .subscribe(
                    i -> {
                        consumedLabel.setText(i.toString());
                        consumedSeries.getData().add(new XYChart.Data<>(System.currentTimeMillis(), i));
                    },
                    Throwable::printStackTrace,
                    () -> startButton.setDisable(false)
            );
}
 

/**
 * Quickly produce numbers
 */
@Override
public Flowable<NumbersProto.Number> sendNumbers(Single<NumbersProto.HowMany> request) {
    // Fork the response flowable using share()
    Flowable<Integer> numbers = request
            // Extract request
            .map(NumbersProto.HowMany::getNumber)
            // Process request
            .flatMapPublisher(i -> Flowable.range(0, i))
            .share();

    // One fork updates the UI
    numbers.observeOn(JavaFxScheduler.platform())
            .subscribe(i -> {
                producedLabel.setText(i.toString());
                producedSeries.getData().add(new XYChart.Data<>(System.currentTimeMillis(), i));
            });

    // Other fork returns the number stream
    return numbers.map(BackpressureController::protoNum);
}
 

@Override
protected void configureBindings() {
	disposables.add(JavaFxObservable.changesOf(canvas.getViews().getChildren())
		.observeOn(JavaFxScheduler.platform())
		.subscribe(next -> setUpCursorOnShapeView(next), ex -> BadaboomCollector.INSTANCE.add(ex)));

	bindDnD2Select();
	bindPressureToSelectShape();
	bindDnDTranslate();
	dbleClickToInitTextSetter();

	shortcutBinder()
		.toProduce(() -> new SelectShapes(canvas.getDrawing()))
		.on(canvas)
		.with(KeyCode.A, OperatingSystem.getControlKey())
		.first(c -> c.getShapes().addAll(canvas.getDrawing().getShapes()))
		.bind();

	shortcutBinder()
		.toProduce(() -> new UpdateToGrid(canvas.getMagneticGrid(), canvas.getDrawing().getSelection().duplicateDeep(false)))
		.on(canvas)
		.with(KeyCode.U, OperatingSystem.getControlKey())
		.when(i -> prefs.isMagneticGrid())
		.bind();
}
 

public void bindTo(final ObservableMeshProgress meshProgress)
{
	unbind();
	this.meshProgress = meshProgress;
	if (this.meshProgress != null)
	{
		this.disposable = JavaFxObservable
				.invalidationsOf(this.meshProgress)
				.throttleLast(updateIntervalMsec, TimeUnit.MILLISECONDS)
				.observeOn(JavaFxScheduler.platform())
				.subscribe(val -> {
					final int numTasks = meshProgress.getNumTasks();
					final int numCompletedTasks = meshProgress.getNumCompletedTasks();

					if (numTasks == 0)
						setProgress(0.0); // hide progress bar when there is nothing to do
					else if (numCompletedTasks <= 0)
						setProgress(1e-7); // displays an empty progress bar
					else if (numCompletedTasks >= numTasks) {
						setStyle(ProgressStyle.FINISHED);
						setProgress(1.0);
					}
					else {
						setStyle(ProgressStyle.IN_PROGRESS);
						setProgress((double) numCompletedTasks / numTasks);
					}

					statusToolTip.setText(numCompletedTasks + "/" + numTasks);
				});
	}
}
 

/**
 * Build the UI and wire up event handlers
 */
@Override
public void start(Stage primaryStage) {
    primaryStage.setTitle("Reactive Chat - " + author);
    Scene scene = buildScene();



    /* ******************************
     * Subscribe to incoming messages
     * ******************************/
    disposables.add(Single
            // Trigger
            .just(Empty.getDefaultInstance())
            .subscribeOn(Schedulers.io())
            // Invoke
            .as(stub::getMessages)
            .map(this::fromMessage)
            // Execute
            .observeOn(JavaFxScheduler.platform())
            .subscribe(messages::appendText));



    /* *************************
     * Publish outgoing messages
     * *************************/
    disposables.add(JavaFxObservable
            // Trigger
            .actionEventsOf(send)
            .subscribeOn(Schedulers.io())
            // Invoke
            .map(x -> message.getText())
            .map(this::toMessage)
            .flatMapSingle(stub::postMessage)
            // Execute
            .observeOn(JavaFxScheduler.platform())
            .subscribe(ignore -> message.clear()));



    /* *********************************
     * Publish arrival/departure message
     * *********************************/
    disposables.add(WindowEventSource
            // Trigger
            .fromWindowEvents(primaryStage, WindowEvent.ANY)
            .subscribeOn(Schedulers.io())
            .filter(event -> event.getEventType().equals(WindowEvent.WINDOW_SHOWING) |
                    event.getEventType().equals(WindowEvent.WINDOW_HIDING))
            // Invoke
            .map(event -> event.getEventType().equals(WindowEvent.WINDOW_SHOWING) ? "joined" : "left")
            .map(this::toMessage)
            .flatMapSingle(stub::postMessage)
            // Execute
            .subscribe());



    primaryStage.setScene(scene);
    primaryStage.show();
}
 

/**
 * Creates the canvas.
 */
@Inject
public Canvas(final PreferencesService prefs, final ViewFactory viewFactory) {
	super();
	this.viewFactory = Objects.requireNonNull(viewFactory);
	modified = false;
	drawing = ShapeFactory.INST.createDrawing();
	zoom = new SimpleDoubleProperty(1d);
	tempView = Optional.empty();
	page = new PageView(prefs, getOrigin());
	magneticGrid = new MagneticGrid(this, prefs);
	disposables = new ArrayList<>();

	widgetsPane = new Group();
	shapesPane = new Group();
	shapesToViewMap = new HashMap<>();
	selectionBorder = new Rectangle();
	ongoingSelectionBorder = new Rectangle();

	widgetsPane.setFocusTraversable(false);
	ongoingSelectionBorder.setFocusTraversable(false);
	ongoingSelectionBorder.setMouseTransparent(true);
	ongoingSelectionBorder.setFill(null);
	ongoingSelectionBorder.setStroke(Color.GRAY);
	ongoingSelectionBorder.setStrokeLineCap(StrokeLineCap.BUTT);
	ongoingSelectionBorder.getStrokeDashArray().addAll(7d, 7d);

	getChildren().add(page);
	getChildren().add(magneticGrid);
	getChildren().add(shapesPane);
	getChildren().add(widgetsPane);
	widgetsPane.getChildren().add(selectionBorder);
	widgetsPane.getChildren().add(ongoingSelectionBorder);
	widgetsPane.relocate(ORIGIN.getX(), ORIGIN.getY());
	shapesPane.relocate(ORIGIN.getX(), ORIGIN.getY());

	defineShapeListToViewBinding();

	selectionBorder.setFocusTraversable(false);
	selectionBorder.setVisible(false);
	selectionBorder.setFill(null);
	selectionBorder.setStroke(Color.GRAY);
	selectionBorder.setStrokeLineCap(StrokeLineCap.BUTT);
	selectionBorder.getStrokeDashArray().addAll(7d, 7d);

	// Bloody key shortcuts. To work the canvas must grab the focus
	// Must be a MOUSE_CLICKED, not a MOUSE_PRESSED, do not know why...
	io.github.interacto.jfx.binding.Bindings
		.anonCmdBinder(() -> requestFocus())
		.usingInteraction(Click::new)
		.on(this)
		.bind();

	// Instead of triggering the update on each change, wait for 20 ms
	disposables.add(JavaFxObservable.<ObservableList<Shape>>changesOf(drawing.getSelection().getShapes())
		.throttleLast(20, TimeUnit.MILLISECONDS)
		.observeOn(JavaFxScheduler.platform())
		.subscribe(next -> updateSelectionBorders(), ex -> BadaboomCollector.INSTANCE.add(ex)));

	disposables.add(CommandsRegistry.getInstance().commands()
		.filter(c -> c instanceof Modifying || c instanceof Undo || c instanceof Redo)
		.subscribe(c -> update()));

	shapesPane.setFocusTraversable(false);

	prefWidthProperty().bind(Bindings.createDoubleBinding(() -> margins * 2d + prefs.getPage().getWidth() * Shape.PPC, prefs.pageProperty()));
	prefHeightProperty().bind(Bindings.createDoubleBinding(() -> margins * 2d + prefs.getPage().getHeight() * Shape.PPC, prefs.pageProperty()));
}