Wrap a connectable observable.
@param source The source
@param instrumentations The instrumentations
@param <T> The type
@return The wrapped subscriber
778364
static <T> ConnectableObservable<T> wrap(
            ConnectableObservable<T> source, Collection<ReactiveInstrumenter> instrumentations) {
        return new RxInstrumentedConnectableObservable<>(source, instrumentations);
    }
572569
function subscribe(observable, observer, options) {
	if (typeof observable !== 'object' || !observable)
		throw new TypeError('observable argument must be an Object');
	if (typeof observable.on !== 'function')
		throw new TypeError('observable.on must be a Function');
	if (typeof observer !== 'object' || !observer)
		throw new TypeError('observer argument must be an Object');

	const { masterHandler, messageTypes, queueName } = options;
	if (masterHandler && typeof masterHandler !== 'function')
		throw new TypeError('masterHandler parameter, when provided, must be a Function');
	if (queueName && typeof observable.queue !== 'function')
		throw new TypeError('observable.queue, when queueName is specified, must be a Function');

	const subscribeTo = messageTypes
		|| observer.handles
		|| Object.getPrototypeOf(observer).constructor.handles;
	if (!Array.isArray(subscribeTo))
		throw new TypeError('either options.messageTypes, observer.handles or ObserverType.handles is required');

	unique(subscribeTo).forEach(messageType => {
		const handler = masterHandler || getHandler(observer, messageType);
		if (!handler)
			throw new Error(`'${messageType}' handler is not defined or not a function`);

		if (queueName)
			observable.queue(queueName).on(messageType, handler.bind(observer));
		else
			observable.on(messageType, handler.bind(observer));
	});
}
603428
public static <T> BeanObservablesDecorator<T> decorate(Observable<T> observable)
	{
		return new BeanObservablesDecorator<T>(observable);
	}
992765
public void subscribe(final Class<? extends T> clazz, final EventHandler<? extends T> handler) {
    lock.writeLock().lock();
    try {
      List<EventHandler<? extends T>> list = clazzToListOfHandlersMap.get(clazz);
      if (list == null) {
        list = new LinkedList<EventHandler<? extends T>>();
        clazzToListOfHandlersMap.put(clazz, list);
      }
      list.add(handler);
    } finally {
      lock.writeLock().unlock();
    }
  }
1158674
@NonNull
    public ObservableDataBuilder<T> prepends(@Nullable Observable<? extends Collection<? extends T>> prepends) {
        mPrepends = prepends;
        return this;
    }
479821
static Completable wrap(
            CompletableSource source, Collection<ReactiveInstrumenter> instrumentations) {
        if (source instanceof Callable) {
            return new RxInstrumentedCallableCompletable<>(source, instrumentations);
        }
        return new RxInstrumentedCompletable(source, instrumentations);
    }
190554
static <T> ParallelFlowable<T> wrap(
            ParallelFlowable<T> source, Collection<ReactiveInstrumenter> instrumentations) {
        return new RxInstrumentedParallelFlowable<>(source, instrumentations);
    }
228867
protected <E extends Event> Listener subscribeEvent(Class<E> eventClass, final Class<? extends Listener<E>> listenerClass) {
        Listener<E> listener = locator.createAndInitialize(listenerClass);
        subscribe(eventClass, listener);
        return listener;
    }
621677
public void addSubscribedSource(Source source) {
        String sourceId = CollectorManagerUtils.getSourceId(source);
        //If present in pending subscription list remove it.
        pendingSubscriptions.remove(sourceId);
        subscribedSources.add(sourceId);
    }
1186505
public void addObservableListener(ObservableListener<T> listener)
	{
		checkNotNull(listener, "listener cannot be null");

		listeners.add(ObservableListener.class, listener);
	}
153563
def addObserver(self, observer):
        """"""
        Observable.addObserver(self, observer)

        # If self.startOnDemand is True, the reader monitoring
        # thread only runs when there are observers.
        if self.startOnDemand:
            if 0 < self.countObservers():
                if not self.rmthread:
                    self.rmthread = ReaderMonitoringThread(
                        self,
                        self.readerProc, self.period)

                    # start reader monitoring thread in another thread to
                    # avoid a deadlock; addObserver and notifyObservers called
                    # in the ReaderMonitoringThread run() method are
                    # synchronized
                    try:
                        # Python 3.x
                        import _thread
                        _thread.start_new_thread(self.rmthread.start, ())
                    except:
                        # Python 2.x
                        import thread
                        thread.start_new_thread(self.rmthread.start, ())
        else:
            observer.update(self, (self.rmthread.readers, []))
1331961
public MethodDoc wrap(MethodDoc source) {
		if (source == null || source instanceof Proxy<?> || !(source instanceof MethodDocImpl)) {
			return source;
		}
		return new MethodDocWrapper((MethodDocImpl) source);
	}
1176133
protected <E extends Event> void subscribeEvent(Class<E> eventClass, final Listener<E> listener) {
        locator.inject(listener);
        locator.postConstruct(listener);
        subscribe(eventClass, listener);
    }
1130599
function observable(options = {}) {
    const observableDecorator = function observableDecorator(elementDescriptor) {
        return getElementDescriptorForProp(
            elementDescriptor.key,
            null,
            ("enumerable" in options) ? options.enumerable : true,
            elementDescriptor.initializer || elementDescriptor.descriptor.get
        );
    };

    if (options && options.key) {
        return observableDecorator(options);
    } else {
        return observableDecorator;
    }
}
1172055
public Closeable onSubscribeChannelEvent(final Consumer<BitfinexStreamSymbol> listener) {
        subscribeChannelConsumers.offer(listener);
        return () -> subscribeChannelConsumers.remove(listener);
    }
264920
public void addSubscriber(Handler handler) {
        String handlerId = CollectorManagerUtils.getHandlerId(handler);

        //temporary exception for audit Source  //must change later
        if (source.getSourceName().trim().equals("audit") && subscribers.isEmpty()) {
            //First subscriber, assign a buffer.
            bufferMgr = new BufferManagerImpl(BUFFER_SIZE, sourceId, false);
            //Inform the source that a buffer is now available
            //and it can start sending events to this buffer.
            source.setBufferManager(this.bufferMgr);
        }
        subscribers.add(handlerId);
        /*
         * Inform the handler that this source/buffer/conduit is now available:
         * Synchronous Handler - Add handler as synchronous handler to the Buffer/Conduit's sync handler set
         * Asynchronous Handler - Add handler as asynchronous handler into the Buffer/Conduit
         *
         */
        if (handler instanceof SynchronousHandler) {
            bufferMgr.addSyncHandler((SynchronousHandler) handler);
        } else {
            bufferMgr.addHandler(handlerId);
        }

    }
1318443
function using(resourceFactory, observableFactory) {
    return new _Observable__WEBPACK_IMPORTED_MODULE_0__["Observable"](function (subscriber) {
        var resource;
        try {
            resource = resourceFactory();
        }
        catch (err) {
            subscriber.error(err);
            return undefined;
        }
        var result;
        try {
            result = observableFactory(resource);
        }
        catch (err) {
            subscriber.error(err);
            return undefined;
        }
        var source = result ? Object(_from__WEBPACK_IMPORTED_MODULE_1__["from"])(result) : _empty__WEBPACK_IMPORTED_MODULE_2__["EMPTY"];
        var subscription = source.subscribe(subscriber);
        return function () {
            subscription.unsubscribe();
            if (resource) {
                resource.unsubscribe();
            }
        };
    });
}