[do not yet merge] Refactor EventSource

src/main/java/rx/observables/SwingObservable.java

@@ -21,30 +21,52 @@
 import java.util.Set;
 
 import javax.swing.AbstractButton;
+import javax.swing.JSlider;
+import javax.swing.JSpinner;
 import javax.swing.ListSelectionModel;
 import javax.swing.SwingUtilities;
+import javax.swing.event.ChangeEvent;
+import javax.swing.event.ChangeListener;
 import javax.swing.event.DocumentEvent;
 import javax.swing.event.ListSelectionEvent;
 import javax.swing.text.Document;
 
 import rx.Observable;
+import rx.Subscriber;
+import rx.Observable.OnSubscribe;
 import rx.functions.Func1;
+import rx.schedulers.SwingScheduler;
 import rx.swing.sources.*;
 
 /**
  * Allows creating observables from various sources specific to Swing. 
  */
 public enum SwingObservable { ; // no instances
 
+    private static <T> Observable<T> create(OnSubscribe<T> f) {
+        return Observable.create(f)
+                .subscribeOn(SwingScheduler.getInstance())
+                .unsubscribeOn(SwingScheduler.getInstance());
+    }
+
     /**
      * Creates an observable corresponding to a Swing button action.
      * 
      * @param button 
      *            The button to register the observable for.
      * @return Observable of action events.
      */
-    public static Observable<ActionEvent> fromButtonAction(AbstractButton button) {
-        return AbstractButtonSource.fromActionOf(button);
+    public static Observable<ActionEvent> fromButtonAction(final AbstractButton button) {
+        return create(new ActionEventSource() {
+            @Override
+            public void addListenerToComponent(ActionListener listener) {
+                button.addActionListener(listener);
+            }
+            @Override
+            public void removeListenerFromComponent(ActionListener listener) {
+                button.removeActionListener(listener);
+            }
+        });
     }
 
     /**
@@ -55,7 +77,7 @@ public static Observable<ActionEvent> fromButtonAction(AbstractButton button) {
      * @return Observable of key events.
      */
     public static Observable<KeyEvent> fromKeyEvents(Component component) {
-        return KeyEventSource.fromKeyEventsOf(component);
+        return create(new KeyEventSource(component));
     }
 
     /**
@@ -82,7 +104,7 @@ public Boolean call(KeyEvent event) {
      * @return Observable of currently pressed keys.
      */
     public static Observable<Set<Integer>> fromPressedKeys(Component component) {
-        return KeyEventSource.currentlyPressedKeysOf(component);
+        return KeyEventSource.currentlyPressedKeys(fromKeyEvents(component));
     }
 
     /**
@@ -136,7 +158,7 @@ public static Observable<MouseWheelEvent> fromMouseWheelEvents(Component compone
      * @return Observable of component events.
      */
     public static Observable<ComponentEvent> fromComponentEvents(Component component) {
-        return ComponentEventSource.fromComponentEventsOf(component);
+        return create(new ComponentEventSource(component));
     }
 
     /**
@@ -158,7 +180,7 @@ public static Observable<FocusEvent> fromFocusEvents(Component component) {
      * @return Observable emitting the current size of the given component after each resize event.
      */
     public static Observable<Dimension> fromResizing(Component component) {
-        return ComponentEventSource.fromResizing(component);
+        return ComponentEventSource.resizing(fromComponentEvents(component));
     }
 
     /**
@@ -169,7 +191,7 @@ public static Observable<Dimension> fromResizing(Component component) {
      * @return Observable emitting the item events for the given itemSelectable.
      */
     public static Observable<ItemEvent> fromItemEvents(ItemSelectable itemSelectable) {
-        return ItemEventSource.fromItemEventsOf(itemSelectable);
+        return create(new ItemEventSource(itemSelectable));
     }
 
     /**
@@ -180,7 +202,7 @@ public static Observable<ItemEvent> fromItemEvents(ItemSelectable itemSelectable
      * @return Observable emitting the an item event whenever the given itemSelectable is selected.
      */
     public static Observable<ItemEvent> fromItemSelectionEvents(ItemSelectable itemSelectable) {
-        return ItemEventSource.fromItemEventsOf(itemSelectable).filter(new Func1<ItemEvent, Boolean>() {
+        return fromItemEvents(itemSelectable).filter(new Func1<ItemEvent, Boolean>() {
             @Override
             public Boolean call(ItemEvent event) {
                 return event.getStateChange() == ItemEvent.SELECTED;
@@ -196,7 +218,7 @@ public Boolean call(ItemEvent event) {
      * @return Observable emitting the an item event whenever the given itemSelectable is deselected.
      */
     public static Observable<ItemEvent> fromItemDeselectionEvents(ItemSelectable itemSelectable) {
-        return ItemEventSource.fromItemEventsOf(itemSelectable).filter(new Func1<ItemEvent, Boolean>() {
+        return fromItemEvents(itemSelectable).filter(new Func1<ItemEvent, Boolean>() {
             @Override
             public Boolean call(ItemEvent event) {
                 return event.getStateChange() == ItemEvent.DESELECTED;
@@ -284,6 +306,46 @@ public Boolean call(DocumentEvent event) {
         });
     }
 
+    /**
+     * Creates an observable corresponding to JSlider changes.
+     * 
+     * @param component
+     *            The slider to register the observable for.
+     * @return Observable of change events.
+     */
+    public static Observable<ChangeEvent> fromChangeEvents(final JSlider component) {
+        return create(new ChangeEventSource() {
+            @Override
+            protected void addListenerToComponent(ChangeListener listener) {
+                component.addChangeListener(listener);
+            }
+            @Override
+            protected void removeListenerFromComponent(ChangeListener listener) {
+                component.removeChangeListener(listener);
+            }
+        });
+    }
+
+    /**
+     * Creates an observable corresponding to JSpinner changes.
+     * 
+     * @param component
+     *            The spinner to register the observable for.
+     * @return Observable of change events.
+     */
+    public static Observable<ChangeEvent> fromChangeEvents(final JSpinner component) {
+        return create(new ChangeEventSource() {
+            @Override
+            protected void addListenerToComponent(ChangeListener listener) {
+                component.addChangeListener(listener);
+            }
+            @Override
+            protected void removeListenerFromComponent(ChangeListener listener) {
+                component.removeChangeListener(listener);
+            }
+        });
+    }
+
     /**
      * Check if the current thead is the event dispatch thread.
      * 

src/main/java/rx/swing/sources/ActionEventSource.java

@@ -0,0 +1,39 @@
+/**
+ * Copyright 2014 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.swing.sources;
+
+import rx.Subscriber;
+
+import java.awt.event.ActionEvent;
+import java.awt.event.ActionListener;
+
+/**
+ * Note: There's no common supertype of all components which have add/removeActionListener
+ * methods, so we cannot implement add/removeListenerForComponent methods here, so this
+ * class has to remain abstract.
+ */
+public abstract class ActionEventSource extends EventSource<ActionEvent, ActionListener> {
+
+    @Override
+    public ActionListener createListenerFor(final Subscriber<? super ActionEvent> subscriber) {
+        return new ActionListener() {
+            @Override
+            public void actionPerformed(ActionEvent e) {
+                subscriber.onNext(e);
+            }
+        };
+    }
+}

src/main/java/rx/swing/sources/ChangeEventSource.java

@@ -0,0 +1,39 @@
+/**
+ * Copyright 2014 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.swing.sources;
+
+import rx.Subscriber;
+
+import javax.swing.event.ChangeEvent;
+import javax.swing.event.ChangeListener;
+
+/**
+ * Note: There's no common supertype of all components which have add/removeChangeListener
+ * methods, so we cannot implement add/removeListenerForComponent methods here, so this
+ * class has to remain abstract.
+ */
+public abstract class ChangeEventSource extends EventSource<ChangeEvent, ChangeListener> {
+
+    @Override
+    public ChangeListener createListenerFor(final Subscriber<? super ChangeEvent> subscriber) {
+        return new ChangeListener() {
+            @Override
+            public void stateChanged(ChangeEvent e) {
+                subscriber.onNext(e);
+            }
+        };
+    }
+}