Create an observable for this eventType.  A new event bus subscription
is made for each call to the Observables.subscribe().  The subscription
is removed from the underlying EventBus when unsubscribe is called.

@param eventType
@return
975845
public <T> Observable<T> asObservable(final Class<T> eventType) {
        return Observable.create(new OnSubscribe<T>() {
            @Override
            public void call(final Subscriber<? super T> observer) {
                final DynamicSubscriber consumer = new DynamicSubscriber() {
                    @Override
                    public Class<?> getEventType() {
                        return eventType;
                    }
                    
                    @Subscribe
                    public void consume(Object obj) {
                        observer.onNext((T) obj);
                    }
                };
                
                observer.add(BooleanSubscription.create(new Action0() {
                    @Override
                    public void call() {
                        eventBus.unregisterSubscriber(consumer);
                        observer.onCompleted();
                    }
                }));
                
                try {
                    eventBus.registerSubscriber(consumer);
                } catch (InvalidSubscriberException e) {
                    observer.onError(e);
                }
            }
        });
    }
1369297
function fromEventCapture(element, name) {
  return Observable.create((subj) => {
    const handler = function (...args) {
      if (args.length > 1) {
        subj.next(args);
      } else {
        subj.next(args[0] || true);
      }
    };

    element.addEventListener(name, handler, true);
    return new Subscription(() => element.removeEventListener(name, handler, true));
  });
}
74022
public Observable<SwipeEvent> observe() {
    this.swipeListener = createReactiveSwipeListener();
    return Observable.create(new ObservableOnSubscribe<SwipeEvent>() {
      @Override public void subscribe(ObservableEmitter<SwipeEvent> emitter) throws Exception {
        Swipe.this.emitter = emitter;
      }
    });
  }
1324157
public static <T> Observable<T> toObservable(EventPublisher<T> eventPublisher) {
        PublishSubject<T> publishSubject = PublishSubject.create();
        Subject<T> serializedSubject = publishSubject.toSerialized();
        eventPublisher.onEvent(serializedSubject::onNext);
        return serializedSubject;
    }
885711
@Override
  public <T> ServiceBuilder service(Class<T> type)
  {
    Key<?> key = Key.of(type, ServiceImpl.class);

    return service(key, type);
  }
1048232
public static <T> Observable<T> observable(Publisher<T> publisher) {
        return Flowable.fromPublisher(publisher).toObservable();
    }
456405
protected <E extends Event> void subscribeSystemEvent(Class<E> eventClass, final Listener<E> listener) {
        locator.inject(listener);
        locator.postConstruct(listener);
        SystemEventBus.subscribe(eventClass, listener);
    }
196161
protected final <U extends GenericEvent> void bind(
      EventBus eventBus,
      List<HandlerRegistration> registrations,
      Class<U> type,
      GenericEventHandler handler) {
    registrations.add(eventBus.addHandler(GenericEventType.getTypeOf(type), handler));
  }
1026823
@SuppressWarnings("unchecked")
    public static <T> ConnectableObservable<T> createFrom(ObservableSource<? extends T> source) {
        ObservableSource<T> sourceCast = (ObservableSource<T>) source;
        return create(sourceCast, DEFAULT_UNBOUNDED_FACTORY);
    }
92221
public Webhook addEventTypes(final String eventType) {
		if (this.eventTypes == null) {
			this.eventTypes = new ArrayList<String>();
		}
		this.eventTypes.add(eventType);
		return this;
	}
984993
public AutoCloseable registerHandler(
      final Class<? extends T> messageType,
      final EventHandler<RemoteMessage<? extends T>> theHandler) {

    this.msgTypeToHandlerMap.put(messageType, theHandler);

    LOG.log(Level.FINER, "Add handler for class: {0}", messageType.getCanonicalName());

    return new SubscriptionHandler<>(messageType, this.unsubscribeClass);
  }
1356349
public <T> List<Subscriber<T>> unsubscribeAll(Class<T> messageType) {
		return subscribeStrategy.unsubscribeAll(mapping, messageType);
	}
531427
protected <E extends Event> Listener subscribeSystemEvent(Class<E> eventClass, final Class<? extends Listener<E>> listenerClass) {
        Listener<E> listener = locator.createAndInitialize(listenerClass);
        SystemEventBus.subscribe(eventClass, listener);
        return listener;
    }
1004911
public static <T> ServiceFuture<T> fromResponse(final Observable<ServiceResponse<T>> observable, final ServiceCallback<T> callback) {
        final ServiceFuture<T> serviceFuture = new ServiceFuture<>();
        serviceFuture.subscription = observable
                .last()
                .subscribe(new Action1<ServiceResponse<T>>() {
                    @Override
                    public void call(ServiceResponse<T> t) {
                        if (callback != null) {
                            callback.success(t.body());
                        }
                        serviceFuture.set(t.body());
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        if (callback != null) {
                            callback.failure(throwable);
                        }
                        serviceFuture.setException(throwable);
                    }
                });
        return serviceFuture;
    }
449360
function (eventType, evt) {
    if (!this.eventListeners) {
      this.eventListeners = {};
    }
    if (!evt) {
      evt = {};
    }
    evt.type = eventType;
    if (!evt.source) {
      evt.source = this;
    }
    var events = this.eventListeners[eventType];
    if (events) {
      var len = events.length;
      for (var i = 0; i < len; i++) {
        events[i].handler.apply(this, [evt]);
      }
    }
    return this;
  }
1245686
public Observable<List<EventTypeInner>> listEventTypesAsync(String topicTypeName) {
        return listEventTypesWithServiceResponseAsync(topicTypeName).map(new Func1<ServiceResponse<List<EventTypeInner>>, List<EventTypeInner>>() {
            @Override
            public List<EventTypeInner> call(ServiceResponse<List<EventTypeInner>> response) {
                return response.body();
            }
        });
    }
28908
def get_subscriber_queue(self, event_types=None):
        

        try:
            self.started_queue.get(timeout=1)
            raise RuntimeError('Cannot create a new subscriber queue while Exchange is running.')
        except Empty:
            pass

        if event_types is None:
            event_types = EventTypes.ALL

        queue = Queue()
        self.queues[event_types].append(queue)
        return queue