test_class.file: rxjava/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java
test_class.identifier: FlowablePublishTest
test_case.body: @Test
      public void selectorInnerError() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          pp.publish(new Function<Flowable<Integer>, Flowable<Integer>>() {
              @Override
              public Flowable<Integer> apply(Flowable<Integer> v) throws Exception {
                  return Flowable.error(new TestException());
              }
          })
          .test()
          .assertFailure(TestException.class);
  
          assertFalse(pp.hasSubscribers());
      }
test_case.class_method_signature: FlowablePublishTest.selectorInnerError()
test_case.full_signature: @Test public void selectorInnerError()
test_case.identifier: selectorInnerError
test_case.invocations: create assertFailure test publish error assertFalse hasSubscribers
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void selectorInnerError()
focal_class.declarator: CANCELLED = Long.MIN_VALUE
focal_class.modifier: static final
focal_class.original_string: static final long CANCELLED = Long.MIN_VALUE;
focal_class.type: long
focal_class.var_name: CANCELLED
focal_class.declarator: source
focal_class.modifier: final
focal_class.original_string: final Flowable<T> source;
focal_class.type: Flowable<T>
focal_class.var_name: source
focal_class.declarator: current
focal_class.modifier: final
focal_class.original_string: final AtomicReference<PublishSubscriber<T>> current;
focal_class.type: AtomicReference<PublishSubscriber<T>>
focal_class.var_name: current
focal_class.declarator: bufferSize
focal_class.modifier: final
focal_class.original_string: final int bufferSize;
focal_class.type: int
focal_class.var_name: bufferSize
focal_class.declarator: onSubscribe
focal_class.modifier: final
focal_class.original_string: final Publisher<T> onSubscribe;
focal_class.type: Publisher<T>
focal_class.var_name: onSubscribe
focal_class.file: rxjava/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java
focal_class.identifier: FlowablePublish
focal_class.interfaces: implements HasUpstreamPublisher<T>
focal_class.class_method_signature: FlowablePublish.create(Flowable<T> source, final int bufferSize)
focal_class.full_signature: public static ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize)
focal_class.identifier: create
focal_class.modifiers: public static
focal_class.parameters: (Flowable<T> source, final int bufferSize)
focal_class.return: ConnectableFlowable<T>
focal_class.signature: ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize)
focal_class.class_method_signature: FlowablePublish.FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<PublishSubscriber<T>> current, int bufferSize)
focal_class.full_signature: private  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<PublishSubscriber<T>> current, int bufferSize)
focal_class.identifier: FlowablePublish
focal_class.modifiers: private
focal_class.parameters: (Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<PublishSubscriber<T>> current, int bufferSize)
focal_class.signature:  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
              final AtomicReference<PublishSubscriber<T>> current, int bufferSize)
focal_class.class_method_signature: FlowablePublish.source()
focal_class.full_signature: @Override public Publisher<T> source()
focal_class.identifier: source
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: Publisher<T>
focal_class.signature: Publisher<T> source()
focal_class.class_method_signature: FlowablePublish.subscribeActual(Subscriber<? super T> s)
focal_class.full_signature: @Override protected void subscribeActual(Subscriber<? super T> s)
focal_class.identifier: subscribeActual
focal_class.modifiers: @Override protected
focal_class.parameters: (Subscriber<? super T> s)
focal_class.return: void
focal_class.signature: void subscribeActual(Subscriber<? super T> s)
focal_class.class_method_signature: FlowablePublish.connect(Consumer<? super Disposable> connection)
focal_class.full_signature: @Override public void connect(Consumer<? super Disposable> connection)
focal_class.identifier: connect
focal_class.modifiers: @Override public
focal_class.parameters: (Consumer<? super Disposable> connection)
focal_class.return: void
focal_class.signature: void connect(Consumer<? super Disposable> connection)
focal_class.superclass: extends ConnectableFlowable<T>
focal_method.body: public static <T> ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize) {
          // the current connection to source needs to be shared between the operator and its onSubscribe call
          final AtomicReference<PublishSubscriber<T>> curr = new AtomicReference<PublishSubscriber<T>>();
          Publisher<T> onSubscribe = new FlowablePublisher<T>(curr, bufferSize);
          return RxJavaPlugins.onAssembly(new FlowablePublish<T>(onSubscribe, source, curr, bufferSize));
      }
focal_method.class_method_signature: FlowablePublish.create(Flowable<T> source, final int bufferSize)
focal_method.full_signature: public static ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize)
focal_method.identifier: create
focal_method.invocations: onAssembly
focal_method.modifiers: public static
focal_method.parameters: (Flowable<T> source, final int bufferSize)
focal_method.return: ConnectableFlowable<T>
focal_method.signature: ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize)