本文整理了Java中reactor.core.publisher.Flux.ofType()方法的一些代码示例，展示了Flux.ofType()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Flux.ofType()方法的具体详情如下：
包路径：reactor.core.publisher.Flux
类名称：Flux
方法名：ofType

Flux.ofType介绍

[英]Evaluate each accepted value against the given Class type. If the a value matches the type, it is passed into the resulting Flux. Otherwise the value is ignored and a request of 1 is emitted.
[中]根据给定的类类型计算每个接受的值。如果a值与类型匹配，则将其传递到生成的通量中。否则将忽略该值并发出1的请求。

代码示例

代码示例来源：origin: codecentric/spring-boot-admin

public void start() {
  subscription = Flux.from(publisher)
            .log(log.getName(), Level.FINEST)
            .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType))
            .ofType(eventType)
            .cast(eventType)
            .compose(this::handle)
            .onErrorContinue((ex, value) -> log.warn("Unexpected error while handling {}", value, ex))
            .subscribe();
}

代码示例来源：origin: reactor/reactor-core

@Test(expected = NullPointerException.class)
public void sourceNull2() {
  Flux.just(1)
    .ofType(null);
}

代码示例来源：origin: reactor/reactor-core

@Override
public StepVerifier.Step<T> then() {
  return step.consumeSubscriptionWith(s -> {
    //the current subscription
    Scannable lowest = Scannable.from(s);
    //attempt to go back to the leftmost parent to check the Context from its perspective
    Context c = Flux.<Scannable>
        fromStream(lowest.parents())
        .ofType(CoreSubscriber.class)
        .takeLast(1)
        .singleOrEmpty()
        //no parent? must be a ScalaSubscription or similar source
        .switchIfEmpty(
            Mono.just(lowest)
              //unless it is directly a CoreSubscriber, let's try to scan the leftmost, see if it has an ACTUAL
              .map(sc -> (sc instanceof CoreSubscriber) ?
                  sc :
                  sc.scanOrDefault(Scannable.Attr.ACTUAL, Scannable.from(null)))
              //we are ultimately only interested in CoreSubscribers' Context
              .ofType(CoreSubscriber.class)
        )
        .map(CoreSubscriber::currentContext)
        //if it wasn't a CoreSubscriber (eg. custom or vanilla Subscriber) there won't be a Context
        .block();
    this.contextExpectations.accept(c);
  });
}

代码示例来源：origin: reactor/reactor-core

@Override
public DefaultStepVerifierBuilder<T> expectNoAccessibleContext() {
  return consumeSubscriptionWith(sub -> {
        Scannable lowest = Scannable.from(sub);
        Scannable verifierSubscriber = Scannable.from(lowest.scan(Scannable.Attr.ACTUAL));
        Context c = Flux.fromStream(verifierSubscriber.parents())
                .ofType(CoreSubscriber.class)
                .map(CoreSubscriber::currentContext)
                .blockLast();
        if (c != null) {
          throw errorFormatter.assertionError("Expected no accessible Context, got " + c);
        }
      });
}

代码示例来源：origin: reactor/reactor-core

@Test
public void errorOfType() {
  StepVerifier.create(Flux.just(1)
              .ofType(String.class))
        .verifyComplete();
}

代码示例来源：origin: reactor/reactor-core

@Test
public void normalOfType() {
  StepVerifier.create(Flux.just(1)
              .ofType(Number.class))
        .expectNext(1)
        .verifyComplete();
}

代码示例来源：origin: com.aol.cyclops/cyclops-reactor

/**
 * @param clazz
 * @return
 * @see reactor.core.publisher.Flux#ofType(java.lang.Class)
 */
public final <U> Flux<U> ofType(Class<U> clazz) {
  return boxed.ofType(clazz);
}
/**

代码示例来源：origin: reactor/reactor-netty

/**
 * Return a {@link Flux} of {@link HttpContent} containing received chunks
 *
 * @return a {@link Flux} of {@link HttpContent} containing received chunks
 */
default Flux<HttpContent> receiveContent() {
  return receiveObject().ofType(HttpContent.class);
}

代码示例来源：origin: reactor/reactor-netty

/**
   * @return a {@link Flux} of {@link WebSocketFrame} formed frame content
   */
  default Flux<WebSocketFrame> receiveFrames() {
    return receiveObject().ofType(WebSocketFrame.class);
  }
}

代码示例来源：origin: io.projectreactor.netty/reactor-netty

/**
 * Return a {@link Flux} of {@link HttpContent} containing received chunks
 *
 * @return a {@link Flux} of {@link HttpContent} containing received chunks
 */
default Flux<HttpContent> receiveContent() {
  return receiveObject().ofType(HttpContent.class);
}

代码示例来源：origin: com.github.PhilippHeuer/events4j

/**
 * Retrieves a {@link reactor.core.publisher.Flux} of the given event type.
 *
 * @param eventClass the event class to obtain events from
 * @param <E> the eventType
 * @return a new {@link reactor.core.publisher.Flux} of the given eventType
 */
public <E extends Event> Flux<E> onEvent(Class<E> eventClass) {
  return processor.publishOn(scheduler).ofType(eventClass);
}

代码示例来源：origin: io.projectreactor.ipc/reactor-netty

/**
 * Return a {@link Flux} of {@link HttpContent} containing received chunks
 *
 * @return a {@link Flux} of {@link HttpContent} containing received chunks
 */
default Flux<HttpContent> receiveContent() {
  return receiveObject().ofType(HttpContent.class);
}

代码示例来源：origin: io.projectreactor.netty/reactor-netty

/**
   * @return a {@link Flux} of {@link WebSocketFrame} formed frame content
   */
  default Flux<WebSocketFrame> receiveFrames() {
    return receiveObject().ofType(WebSocketFrame.class);
  }
}

代码示例来源：origin: io.projectreactor.ipc/reactor-netty

/**
   * @return a {@link Flux} of {@link WebSocketFrame} formed frame content
   */
  default Flux<WebSocketFrame> receiveFrames() {
    return receiveObject().ofType(WebSocketFrame.class);
  }
}

代码示例来源：origin: io.projectreactor.ipc/reactor-netty

/**
 * Return a {@link Flux} of {@link HttpContent} containing received chunks
 *
 * @return a {@link Flux} of {@link HttpContent} containing received chunks
 */
default Flux<HttpContent> receiveContent(){
  return receiveObject().ofType(HttpContent.class);
}

代码示例来源：origin: io.projectreactor/reactor-test

@Override
public StepVerifier.Step<T> then() {
  return step.consumeSubscriptionWith(s -> {
    //the current subscription
    Scannable lowest = Scannable.from(s);
    //attempt to go back to the leftmost parent to check the Context from its perspective
    Context c = Flux.<Scannable>
        fromStream(lowest.parents())
        .ofType(CoreSubscriber.class)
        .takeLast(1)
        .singleOrEmpty()
        //no parent? must be a ScalaSubscription or similar source
        .switchIfEmpty(
            Mono.just(lowest)
              //unless it is directly a CoreSubscriber, let's try to scan the leftmost, see if it has an ACTUAL
              .map(sc -> (sc instanceof CoreSubscriber) ?
                  sc :
                  sc.scanOrDefault(Scannable.Attr.ACTUAL, Scannable.from(null)))
              //we are ultimately only interested in CoreSubscribers' Context
              .ofType(CoreSubscriber.class)
        )
        .map(CoreSubscriber::currentContext)
        //if it wasn't a CoreSubscriber (eg. custom or vanilla Subscriber) there won't be a Context
        .block();
    this.contextExpectations.accept(c);
  });
}

代码示例来源：origin: io.projectreactor/reactor-test

@Override
public DefaultStepVerifierBuilder<T> expectNoAccessibleContext() {
  return consumeSubscriptionWith(sub -> {
        Scannable lowest = Scannable.from(sub);
        Scannable verifierSubscriber = Scannable.from(lowest.scan(Scannable.Attr.ACTUAL));
        Context c = Flux.fromStream(verifierSubscriber.parents())
                .ofType(CoreSubscriber.class)
                .map(CoreSubscriber::currentContext)
                .blockLast();
        if (c != null) {
          throw errorFormatter.assertionError("Expected no accessible Context, got " + c);
        }
      });
}

代码示例来源：origin: reactor/reactor-netty

x.addHandlerFirst(new HttpClientCodec()))
.receiveObject()
.ofType(DefaultHttpContent.class)
.as(ByteBufFlux::fromInbound)
.asString()