PropertyChangeListenerOnSubscribe.java 文件源码

@Override
@SuppressWarnings("unchecked")
public void call(final Subscriber<? super T> subscriber) {
    final PropertyChangeListener changeListener = new PropertyChangeListener() {
        @Override
        public void propertyChange(PropertyChangeEvent event) {
            if (!subscriber.isUnsubscribed() && property.equals(event.getPropertyName())) {
                subscriber.onNext((T) event.getNewValue());
            }
        }
    };

    propertyChangeSupport.addPropertyChangeListener(changeListener);
    subscriber.add(BooleanSubscription.create(new Action0() {
        @Override
        public void call() {
            propertyChangeSupport.removePropertyChangeListener(changeListener);
        }
    }));
}