Make Subscriptions of SwingObservable thread-safe

rxjava-contrib/rxjava-swing/src/main/java/rx/observables/SwingObservable.java

@@ -25,6 +25,7 @@
 import java.util.Set;
 
 import javax.swing.AbstractButton;
+import javax.swing.SwingUtilities;
 
 import rx.Observable;
 import rx.functions.Func1;
@@ -140,4 +141,15 @@ public static Observable<ComponentEvent> fromComponentEvents(Component component
     public static Observable<Dimension> fromResizing(Component component) {
         return ComponentEventSource.fromResizing(component);
     }
+
+    /**
+     * Check if the current thead is the event dispatch thread.
+     * 
+     * @throws IllegalStateException if the current thread is not the event dispatch thread.
+     */
+    public static void assertEventDispatchThread() {
+        if (!SwingUtilities.isEventDispatchThread()) {
+            throw new IllegalStateException("Need to run in the event dispatch thread, but was " + Thread.currentThread());
+        }
+    }
 }

rxjava-contrib/rxjava-swing/src/main/java/rx/subscriptions/SwingSubscriptions.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.subscriptions;
+
+import javax.swing.SwingUtilities;
+
+import rx.Scheduler.Inner;
+import rx.Subscription;
+import rx.schedulers.SwingScheduler;
+import rx.functions.Action0;
+import rx.functions.Action1;
+
+public final class SwingSubscriptions {
+
+    private SwingSubscriptions() {
+        // no instance
+    }
+
+    /**
+     * Create an Subscription that always runs <code>unsubscribe</code> in the event dispatch thread.
+     * 
+     * @param unsubscribe
+     * @return an Subscription that always runs <code>unsubscribe</code> in the event dispatch thread.
+     */
+    public static Subscription unsubscribeInEventDispatchThread(final Action0 unsubscribe) {
+        return Subscriptions.create(new Action0() {
+            @Override
+            public void call() {
+                if (SwingUtilities.isEventDispatchThread()) {
+                    unsubscribe.call();
+                } else {
+                    SwingScheduler.getInstance().schedule(new Action1<Inner>() {
+                        @Override
+                        public void call(Inner inner) {
+                            unsubscribe.call();
+                        }
+                    });
+                }
+            }
+        });
+    }
+}

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/AbstractButtonSource.java

@@ -15,7 +15,10 @@
  */
 package rx.swing.sources;
 
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import java.awt.event.ActionEvent;
 import java.awt.event.ActionListener;
@@ -26,76 +29,84 @@
 import org.mockito.Matchers;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Action1;
-import rx.subscriptions.Subscriptions;
+import rx.observables.SwingObservable;
+import rx.subscriptions.SwingSubscriptions;
 
 public enum AbstractButtonSource { ; // no instances
 
     /**
      * @see rx.observables.SwingObservable#fromButtonAction
      */
     public static Observable<ActionEvent> fromActionOf(final AbstractButton button) {
-        return Observable.create(new OnSubscribeFunc<ActionEvent>() {
+        return Observable.create(new OnSubscribe<ActionEvent>() {
             @Override
-            public Subscription onSubscribe(final Observer<? super ActionEvent> observer) {
+            public void call(final Subscriber<? super ActionEvent> subscriber) {
+                SwingObservable.assertEventDispatchThread();
                 final ActionListener listener = new ActionListener() {
                     @Override
                     public void actionPerformed(ActionEvent e) {
-                        observer.onNext(e);
+                        subscriber.onNext(e);
                     }
                 };
                 button.addActionListener(listener);
-
-                return Subscriptions.create(new Action0() {
+                subscriber.add(SwingSubscriptions.unsubscribeInEventDispatchThread(new Action0() {
                     @Override
                     public void call() {
                         button.removeActionListener(listener);
                     }
-                });
+                }));
             }
         });
     }
 
     public static class UnitTest {
         @Test
-        public void testObservingActionEvents() {
-            @SuppressWarnings("unchecked")
-            Action1<ActionEvent> action = mock(Action1.class);
-            @SuppressWarnings("unchecked")
-            Action1<Throwable> error = mock(Action1.class);
-            Action0 complete = mock(Action0.class);
-
-            final ActionEvent event = new ActionEvent(this, 1, "command");
-
-            @SuppressWarnings("serial")
-            class TestButton extends AbstractButton {
-                void testAction() {
-                    fireActionPerformed(event);
+        public void testObservingActionEvents() throws Throwable {
+            SwingTestHelper.create().runInEventDispatchThread(new Action0() {
+
+                @Override
+                public void call() {
+                    @SuppressWarnings("unchecked")
+                    Action1<ActionEvent> action = mock(Action1.class);
+                    @SuppressWarnings("unchecked")
+                    Action1<Throwable> error = mock(Action1.class);
+                    Action0 complete = mock(Action0.class);
+
+                    final ActionEvent event = new ActionEvent(this, 1, "command");
+
+                    @SuppressWarnings("serial")
+                    class TestButton extends AbstractButton {
+                        void testAction() {
+                            fireActionPerformed(event);
+                        }
+                    }
+
+                    TestButton button = new TestButton();
+                    Subscription sub = fromActionOf(button).subscribe(action, error, complete);
+
+                    verify(action, never()).call(Matchers.<ActionEvent> any());
+                    verify(error, never()).call(Matchers.<Throwable> any());
+                    verify(complete, never()).call();
+
+                    button.testAction();
+                    verify(action, times(1)).call(Matchers.<ActionEvent> any());
+
+                    button.testAction();
+                    verify(action, times(2)).call(Matchers.<ActionEvent> any());
+
+                    sub.unsubscribe();
+                    button.testAction();
+                    verify(action, times(2)).call(Matchers.<ActionEvent> any());
+                    verify(error, never()).call(Matchers.<Throwable> any());
+                    verify(complete, never()).call();
                 }
-            }
-
-            TestButton button = new TestButton();
-            Subscription sub = fromActionOf(button).subscribe(action, error, complete);
-
-            verify(action, never()).call(Matchers.<ActionEvent>any());
-            verify(error, never()).call(Matchers.<Throwable>any());
-            verify(complete, never()).call();
-
-            button.testAction();
-            verify(action, times(1)).call(Matchers.<ActionEvent>any());
-
-            button.testAction();
-            verify(action, times(2)).call(Matchers.<ActionEvent>any());
-
-            sub.unsubscribe();
-            button.testAction();
-            verify(action, times(2)).call(Matchers.<ActionEvent>any());
-            verify(error, never()).call(Matchers.<Throwable>any());
-            verify(complete, never()).call();
+
+            }).awaitTerminal();
         }
     }
 }

rxjava-contrib/rxjava-swing/src/main/java/rx/swing/sources/ComponentEventSource.java

@@ -15,60 +15,59 @@
  */
 package rx.swing.sources;
 
-import static rx.swing.sources.ComponentEventSource.Predicate.*;
+import static rx.swing.sources.ComponentEventSource.Predicate.RESIZED;
 
 import java.awt.Component;
 import java.awt.Dimension;
 import java.awt.event.ComponentEvent;
 import java.awt.event.ComponentListener;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
-import rx.Observer;
-import rx.Subscription;
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
 import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.observables.SwingObservable;
-import rx.subscriptions.Subscriptions;
+import rx.subscriptions.SwingSubscriptions;
 
 public enum ComponentEventSource { ; // no instances
 
     /**
      * @see rx.observables.SwingObservable#fromComponentEvents
      */
     public static Observable<ComponentEvent> fromComponentEventsOf(final Component component) {
-        return Observable.create(new OnSubscribeFunc<ComponentEvent>() {
+        return Observable.create(new OnSubscribe<ComponentEvent>() {
             @Override
-            public Subscription onSubscribe(final Observer<? super ComponentEvent> observer) {
+            public void call(final Subscriber<? super ComponentEvent> subscriber) {
+                SwingObservable.assertEventDispatchThread();
                 final ComponentListener listener = new ComponentListener() {
                     @Override
                     public void componentHidden(ComponentEvent event) {
-                        observer.onNext(event);
+                        subscriber.onNext(event);
                     }
 
                     @Override
                     public void componentMoved(ComponentEvent event) {
-                        observer.onNext(event);
+                        subscriber.onNext(event);
                     }
 
                     @Override
                     public void componentResized(ComponentEvent event) {
-                        observer.onNext(event);
+                        subscriber.onNext(event);
                     }
 
                     @Override
                     public void componentShown(ComponentEvent event) {
-                        observer.onNext(event);
+                        subscriber.onNext(event);
                     }
                 };
                 component.addComponentListener(listener);
-
-                return Subscriptions.create(new Action0() {
+                subscriber.add(SwingSubscriptions.unsubscribeInEventDispatchThread(new Action0() {
                     @Override
                     public void call() {
                         component.removeComponentListener(listener);
                     }
-                });
+                }));
             }
         });
     }