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

Observable.ofType介绍

[英]Filters the items emitted by an Observable, only emitting those of the specified type.

Scheduler: ofType does not operate by default on a particular Scheduler.
[中]过滤可观察项发出的项，仅发出指定类型的项。
调度器：默认情况下，ofType不会在特定的调度器上运行。

代码示例

代码示例来源：origin: leeowenowen/rxjava-examples

@Override
 public void run() {
  Observable.<Object>//
             just(1, "2", //
               new Exception("abc")).
             ofType(Integer.class).subscribe(new Action1<Integer>() {
   @Override
   public void call(Integer integer) {
    log(integer);
   }
  });
 }
}

代码示例来源：origin: com.github.kmbulebu.dsc/dsc-it100-library

public Labels(Observable<ReadCommand> readObservable, PublishSubject<WriteCommand> writeObservable) {
  readObservable.ofType(BroadcastLabelsCommand.class).subscribe(this);
  writeObservable.onNext(new LabelsRequestCommand());
}

代码示例来源：origin: com.netflix.eureka/eureka2-core

private void installAcknowledgementHandler() {
  connection.getInput()
      .ofType(Acknowledgement.class)
      .subscribe(new Action1<Acknowledgement>() {
        @Override
        public void call(Acknowledgement acknowledgement) {
          PendingAck pending = pendingAckQueue.poll();
          metrics.decrementPendingAckCounter();
          if (pending == null) {
            shutdown(new IllegalStateException("{connection=" + name + "}: unexpected acknowledgment"));
          } else {
            pending.ackSubject.onCompleted();
          }
        }
      });
  schedulerWorker.schedule(ackTimeoutTask, 1, TimeUnit.SECONDS);
}

代码示例来源：origin: spotify/mobius

@Override
 public Observable<E> call(Observable<F> effects) {
  return effects
    .ofType(effectClass)
    .compose(effectHandler)
    .doOnError(onErrorFunction.call(effectHandler));
 }
});

代码示例来源：origin: com.github.kmbulebu.dsc/dsc-it100-library

readObservable.ofType(BaudRateSetCommand.class).subscribe(new Action1<BaudRateSetCommand>() {

代码示例来源：origin: com.github.kmbulebu.dsc/dsc-it100-library

public static void main(String[] args) {
  // Configure for remote RaspberryPI with serial dongle and ser2net setup on port 2000 TCP.
  final IT100 it100 = new IT100(new ConfigurationBuilder().withRemoteSocket("raspberrypi", 2000).build());
  
  try {	
    // Start communicating with IT-100.
    it100.connect();
    
    final Observable<ReadCommand> readObservable = it100.getReadObservable();
    
    // Labels gives us friendly names to our zones.
    final Labels labels = new Labels(readObservable, it100.getWriteObservable());
    
    // Subscribe to Zone opening events
    readObservable.ofType(ZoneOpenCommand.class).subscribe(new Action1<ZoneOpenCommand>() {
      @Override
      public void call(ZoneOpenCommand t1) {
        // Print time and name of zone that opened.
        System.out.println(System.currentTimeMillis() + " " + labels.getZoneLabel(t1.getZone()) + " opened.");
      }
      
    });
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  }
}