code: public void _onNext(IN in) {
          if ( subscribers == null )
              return;
          openRequested--;
          try {
              OUT apply = processor.apply(in);
              if ( apply != null ) {
                  forwardMessage(apply);
              } else {
                  emitRequestNext();
              }
          } catch (Throwable err) {
              err.printStackTrace();
              forwardError(err);
          }
      }
docs: see onError() comment