docs: public static <T, R> Publisher<R> map(Publisher<T> publisher, Function<T, R> mapper) {
          return actual -> publisher.subscribe(new CompletionAwareSubscriber<T>() {
              @Override
              protected void doOnSubscribe(Subscription subscription) {
                  actual.onSubscribe(subscription);
              }
  
              @Override
              protected void doOnNext(T message) {
                  try {
                      R result = Objects.requireNonNull(mapper.apply(message),
                          "The mapper returned a null value.");
                      actual.onNext(result);
                  } catch (Throwable e) {
                      onError(e);
                  }
  
              }
  
              @Override
              protected void doOnError(Throwable t) {
                  actual.onError(t);
              }
  
              @Override
              protected void doOnComplete() {
                  actual.onComplete();
              }
          });
      }