Merge pull request #241 from sialcasa/240_subscriber_on_ui_thread

#240 fix but incl. test case to reproduce the bug.

mvvmfx/src/main/java/de/saxsys/mvvmfx/ViewModel.java

@@ -54,15 +54,16 @@ public interface ViewModel {
 	 *            to be send
 	 */
 	default void publish(String messageName, Object... payload) {
-		if (!Platform.isFxApplicationThread()) {
+		if (Platform.isFxApplicationThread()) {
 			MvvmFX.getNotificationCenter().publish(this, messageName, payload);
 		} else {
-			MvvmFX.getNotificationCenter().publish(this, messageName, payload);
+			Platform.runLater(() -> MvvmFX.getNotificationCenter().publish(this, messageName, payload));
 		}
 	}
 
 	/**
-	 * Subscribe to a notification with a given {@link NotificationObserver}.
+	 * Subscribe to a notification with a given {@link NotificationObserver}. 
+	 * The observer will be invoked on the UI thread.
 	 * 
 	 * @param messageName
 	 *            of the Notification

mvvmfx/src/test/java/de/saxsys/mvvmfx/utils/notifications/ViewModelTest.java

@@ -1,13 +1,25 @@
 package de.saxsys.mvvmfx.utils.notifications;
 
+import de.saxsys.javafx.test.JfxRunner;
 import de.saxsys.mvvmfx.MvvmFX;
 import de.saxsys.mvvmfx.ViewModel;
+import javafx.application.Platform;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 
 import de.saxsys.mvvmfx.utils.notifications.NotificationObserver;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(JfxRunner.class)
 public class ViewModelTest {
 
 	private static final String TEST_NOTIFICATION = "test_notification";
@@ -26,20 +38,44 @@ public void init() {
 		viewModel = new ViewModel() {
 		};
 	}
+
+	@Test
+	public void observerIsCalledFromUiThread() throws InterruptedException, ExecutionException, TimeoutException {
+
+		CompletableFuture<Boolean> future = new CompletableFuture<>();
+
+		// The test doesn't run on the FX thread.
+		assertThat(Platform.isFxApplicationThread()).isFalse();
+
+		viewModel.subscribe(TEST_NOTIFICATION, (key, payload) -> {
+			// the notification is executed on the FX thread.
+			future.complete(Platform.isFxApplicationThread());
+		});
+
+		viewModel.publish(TEST_NOTIFICATION);
+
+
+		final Boolean wasCalledOnUiThread = future.get(1l, TimeUnit.SECONDS);
+
+		assertThat(wasCalledOnUiThread).isTrue();
+	}
 
 
 	@Test
 	public void observerFromOutsideDoesNotReceiveNotifications() {
 		MvvmFX.getNotificationCenter().subscribe(TEST_NOTIFICATION, observer1);
 		viewModel.publish(TEST_NOTIFICATION);
-
+
+		waitForUiThread();
 		Mockito.verify(observer1,  Mockito.never()).receivedNotification(TEST_NOTIFICATION);
 	}
 
 	@Test
 	public void addObserverAndPublish() throws Exception {
 		viewModel.subscribe(TEST_NOTIFICATION, observer1);
 		viewModel.publish(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
+
+		waitForUiThread();
 		Mockito.verify(observer1).receivedNotification(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
 	}
 
@@ -48,11 +84,15 @@ public void addAndRemoveObserverAndPublish() throws Exception {
 		viewModel.subscribe(TEST_NOTIFICATION, observer1);
 		viewModel.unsubscribe(observer1);
 		viewModel.publish(TEST_NOTIFICATION);
+
+		waitForUiThread();
 		Mockito.verify(observer1, Mockito.never()).receivedNotification(TEST_NOTIFICATION);
 
 		viewModel.subscribe(TEST_NOTIFICATION, observer1);
 		viewModel.unsubscribe(TEST_NOTIFICATION, observer1);
 		viewModel.publish(TEST_NOTIFICATION);
+
+		waitForUiThread();
 		Mockito.verify(observer1, Mockito.never()).receivedNotification(TEST_NOTIFICATION);
 	}
 
@@ -62,6 +102,9 @@ public void addMultipleObserverAndPublish() throws Exception {
 		viewModel.subscribe(TEST_NOTIFICATION, observer2);
 		viewModel.subscribe(TEST_NOTIFICATION, observer3);
 		viewModel.publish(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
+
+		waitForUiThread();
+
 		Mockito.verify(observer1).receivedNotification(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
 		Mockito.verify(observer2).receivedNotification(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
 		Mockito.verify(observer3).receivedNotification(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
@@ -75,6 +118,8 @@ public void addMultipleObserverAndRemoveOneAndPublish() throws Exception {
 		viewModel.subscribe(TEST_NOTIFICATION, observer3);
 		viewModel.unsubscribe(observer1);
 		viewModel.publish(TEST_NOTIFICATION, OBJECT_ARRAY_FOR_NOTIFICATION);
+
+		waitForUiThread();
 
 		Mockito.verify(observer1, Mockito.never()).receivedNotification(TEST_NOTIFICATION,
 				OBJECT_ARRAY_FOR_NOTIFICATION);
@@ -83,7 +128,20 @@ public void addMultipleObserverAndRemoveOneAndPublish() throws Exception {
 		Mockito.verify(observer3).receivedNotification(TEST_NOTIFICATION,
 				OBJECT_ARRAY_FOR_NOTIFICATION);
 	}
-
+
+	/**
+	 *	This method is used to wait until the UI thread has done all work that was queued via {@link Platform#runLater(Runnable)}.
+	 */
+	private void waitForUiThread() {
+		CompletableFuture<Void> future = new CompletableFuture<>();
+		Platform.runLater(() -> future.complete(null));
+		try {
+			future.get(1l, TimeUnit.SECONDS);
+		} catch (InterruptedException | ExecutionException | TimeoutException e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
 	private class DummyNotificationObserver implements NotificationObserver {
 		@Override
 		public void receivedNotification(String key, Object... payload) {