code: @CheckReturnValue
      @BackpressureSupport(BackpressureKind.PASS_THROUGH)
      @SchedulerSupport(SchedulerSupport.NONE)
      @SuppressWarnings("unchecked")
      public static <T> Flowable<T> fromPublisher(final Publisher<? extends T> source) {
          if (source instanceof Flowable) {
              return RxJavaPlugins.onAssembly((Flowable<T>)source);
          }
          ObjectHelper.requireNonNull(source, "publisher is null");
  
          return RxJavaPlugins.onAssembly(new FlowableFromPublisher<T>(source));
      }
docs: Converts an arbitrary Reactive-Streams Publisher into a Flowable if not already a
  Flowable.
  <p>
  The {@link Publisher} must follow the
  <a href="https://github.com/reactive-streams/reactive-streams-jvm#reactive-streams">Reactive-Streams specification</a>.
  Violating the specification may result in undefined behavior.
  <p>
  If possible, use {@link #create(FlowableOnSubscribe, BackpressureStrategy)} to create a
  source-like {@code Flowable} instead.
  <p>
  Note that even though {@link Publisher} appears to be a functional interface, it
  is not recommended to implement it through a lambda as the specification requires
  state management that is not achievable with a stateless lambda.
  <dl>
  <dt><b>Backpressure:</b></dt>
  <dd>The operator is a pass-through for backpressure and its behavior is determined by the
  backpressure behavior of the wrapped publisher.</dd>
  <dt><b>Scheduler:</b></dt>
  <dd>{@code fromPublisher} does not operate by default on a particular {@link Scheduler}.</dd>
  </dl>
  @param <T> the value type of the flow
  @param source the Publisher to convert
  @return the new Flowable instance
  @throws NullPointerException if the {@code source} {@code Publisher} is null
  @see #create(FlowableOnSubscribe, BackpressureStrategy)