rx.Observable.ofType()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(2.8k)|赞(0)|评价(0)|浏览(207)

本文整理了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

  1. @Override
  2. public void run() {
  3. Observable.<Object>//
  4. just(1, "2", //
  5. new Exception("abc")).
  6. ofType(Integer.class).subscribe(new Action1<Integer>() {
  7. @Override
  8. public void call(Integer integer) {
  9. log(integer);
  10. }
  11. });
  12. }
  13. }

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

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

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

  1. private void installAcknowledgementHandler() {
  2. connection.getInput()
  3. .ofType(Acknowledgement.class)
  4. .subscribe(new Action1<Acknowledgement>() {
  5. @Override
  6. public void call(Acknowledgement acknowledgement) {
  7. PendingAck pending = pendingAckQueue.poll();
  8. metrics.decrementPendingAckCounter();
  9. if (pending == null) {
  10. shutdown(new IllegalStateException("{connection=" + name + "}: unexpected acknowledgment"));
  11. } else {
  12. pending.ackSubject.onCompleted();
  13. }
  14. }
  15. });
  16. schedulerWorker.schedule(ackTimeoutTask, 1, TimeUnit.SECONDS);
  17. }

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

  1. @Override
  2. public Observable<E> call(Observable<F> effects) {
  3. return effects
  4. .ofType(effectClass)
  5. .compose(effectHandler)
  6. .doOnError(onErrorFunction.call(effectHandler));
  7. }
  8. });

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

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

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

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

相关文章

Observable类方法