Register error handlers before subscribers.

reactfx/src/main/java/org/reactfx/BiEventStream.java

@@ -29,7 +29,7 @@ default Subscription subscribe(Consumer<? super Tuple2<A, B>> subscriber) {
     default Subscription watch(
             BiConsumer<? super A, ? super B> subscriber,
             Consumer<? super Throwable> monitor) {
-        return subscribe(subscriber).and(monitor(monitor));
+        return monitor(monitor).and(subscribe(subscriber));
     }
 
     default Subscription feedTo2(BiEventSink<? super A, ? super B> sink) {

reactfx/src/main/java/org/reactfx/EventStream.java

@@ -55,14 +55,14 @@ public interface EventStream<T> {
 
     /**
      * Convenient method to subscribe to and monitor this stream. Is equivalent
-     * to {@code subscribe(subscriber).and(monitor(monitor))}.
+     * to {@code monitor(monitor).and(subscribe(subscriber))}.
      * @see #subscribe(Consumer)
      * @see #monitor(Consumer)
      */
     default Subscription watch(
             Consumer<? super T> subscriber,
             Consumer<? super Throwable> monitor) {
-        return subscribe(subscriber).and(monitor(monitor));
+        return monitor(monitor).and(subscribe(subscriber));
     }
 
     /**
@@ -715,8 +715,8 @@ default EventStream<Try<T>> materializeErrors() {
         return new LazilyBoundStream<Try<T>>() {
             @Override
             protected Subscription subscribeToInputs() {
-                Subscription s1 = EventStream.this.subscribe(t -> emit(Try.success(t)));
                 Subscription s2 = EventStream.this.monitor(er -> emit(Try.failure(er)));
+                Subscription s1 = EventStream.this.subscribe(t -> emit(Try.success(t)));
                 return s1.and(s2);
             }
         };
@@ -734,8 +734,8 @@ default EventStream<T> handleErrors(Consumer<? super Throwable> handler) {
         return new LazilyBoundStream<T>() {
             @Override
             protected Subscription subscribeToInputs() {
-                Subscription s1 = EventStream.this.subscribe(this::emit);
                 Subscription s2 = EventStream.this.monitor(handler);
+                Subscription s1 = EventStream.this.subscribe(this::emit);
                 return s1.and(s2);
             }
         };

reactfx/src/main/java/org/reactfx/EventStreamBase.java

@@ -46,8 +46,8 @@ protected final void reportError(Throwable thrown) {
      * Overriding this method is a convenient way for subclasses
      * to handle this event.
      *
-     * <p>This method is called after the {@link #newSubscriber(Object)}
-     * method.</p>
+     * <p>This method is called <em>before</em> the
+     * {@link #newSubscriber(Object)} method.</p>
      */
     protected void firstSubscriber() {
         // default implementation is empty
@@ -57,6 +57,9 @@ protected void firstSubscriber() {
      * Called for each new subscriber.
      * Overriding this method is a convenient way for subclasses
      * to handle this event, for example to publish some initial events.
+     *
+     * <p>This method is called <em>after</em> the
+     * {@link #firstSubscriber()} method.</p>
      */
     protected void newSubscriber(S subscriber) {
         // default implementation is empty
@@ -117,10 +120,10 @@ protected final <A, B, C> Subscription subscribeToTri(
 
     public final Subscription subscribe(S subscriber) {
         subscribers = ListHelper.add(subscribers, subscriber);
-        newSubscriber(subscriber);
         if(ListHelper.size(subscribers) == 1) {
             firstSubscriber();
         }
+        newSubscriber(subscriber);
         return () -> unsubscribe(subscriber);
     }
 

reactfx/src/main/java/org/reactfx/EventStreams.java

@@ -168,6 +168,7 @@ protected Subscription subscribeToInputs() {
                 for(Observable dep: dependencies) {
                     dep.addListener(listener);
                 }
+                previousValue = computeValue.get();
 
                 return () -> {
                     for(Observable dep: dependencies) {
@@ -178,9 +179,6 @@ protected Subscription subscribeToInputs() {
 
             @Override
             protected void newSubscriber(Consumer<? super T> subscriber) {
-                if(!isBound()) { // this is the first subscriber
-                    previousValue = computeValue.get();
-                }
                 subscriber.accept(previousValue);
             }
         };

reactfx/src/main/java/org/reactfx/TriEventStream.java

@@ -29,7 +29,7 @@ default Subscription subscribe(Consumer<? super Tuple3<A, B, C>> subscriber) {
     default Subscription watch(
             TriConsumer<? super A, ? super B, ? super C> subscriber,
             Consumer<? super Throwable> monitor) {
-        return subscribe(subscriber).and(monitor(monitor));
+        return monitor(monitor).and(subscribe(subscriber));
     }
 
     default Subscription feedTo3(TriEventSink<? super A, ? super B, ? super C> sink) {

reactfx/src/test/java/org/reactfx/ErrorReportingTest.java

@@ -3,13 +3,16 @@
 import static org.junit.Assert.*;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
 import org.junit.Test;
+import org.reactfx.inhibeans.property.SimpleIntegerProperty;
 
 public class ErrorReportingTest {
 
@@ -48,4 +51,62 @@ public void threadBridgeTest() throws InterruptedException, ExecutionException {
         ex1.shutdown();
         ex2.shutdown();
     }
+    /**
+     * Stream of property values should work normally even when the first value
+     * in the stream (property value at the moment of subscription) produces an
+     * exception. The catch is the first value produced at the very moment of
+     * establishing subscription, so order of initialization matters.
+     */
+    @Test
+    public void property_values_stream_with_faulty_first_value_test() {
+        SimpleIntegerProperty intProperty = new SimpleIntegerProperty(-1);
+        List<String> emitted = new LinkedList<>();
+        List<Throwable> errors = new LinkedList<>();
+
+        EventStreams.valuesOf(intProperty)
+                .map(i -> {
+                    if (i.intValue() < 0) {
+                        throw new IllegalArgumentException("Accepting only positive numbers");
+                    }
+                    return String.valueOf(i);
+                })
+                .handleErrors(errors::add)
+                .subscribe(emitted::add);
+
+        intProperty.set(10);
+        intProperty.set(-2);
+        intProperty.set(0);
+
+        assertEquals(Arrays.asList("10", "0"), emitted);
+        assertEquals(2, errors.size());
+    }
+
+    /**
+     * Variation on
+     * {@link #property_values_stream_with_faulty_first_value_test()} using the
+     * {@link EventStream#watch(java.util.function.Consumer, java.util.function.Consumer)}
+     * method.
+     */
+    @Test
+    public void property_values_stream_with_faulty_first_value_test2() {
+        SimpleIntegerProperty intProperty = new SimpleIntegerProperty(-1);
+        List<String> emitted = new LinkedList<>();
+        List<Throwable> errors = new LinkedList<>();
+
+        EventStreams.valuesOf(intProperty)
+                .map(i -> {
+                    if (i.intValue() < 0) {
+                        throw new IllegalArgumentException("Accepting only positive numbers");
+                    }
+                    return String.valueOf(i);
+                })
+                .watch(emitted::add, errors::add);
+
+        intProperty.set(10);
+        intProperty.set(-2);
+        intProperty.set(0);
+
+        assertEquals(Arrays.asList("10", "0"), emitted);
+        assertEquals(2, errors.size());
+    }
 }