本文整理了Java中rx.Observable.ofType()
方法的一些代码示例,展示了Observable.ofType()
的具体用法。这些代码示例主要来源于Github
/Stackoverflow
/Maven
等平台,是从一些精选项目中提取出来的代码,具有较强的参考意义,能在一定程度帮忙到你。Observable.ofType()
方法的具体详情如下:
包路径:rx.Observable
类名称: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();
}
}
内容来源于网络,如有侵权,请联系作者删除!