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

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

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

Observable.cast介绍

[英]Returns an Observable that emits the items emitted by the source Observable, converted to the specified type.

Scheduler: cast does not operate by default on a particular Scheduler.
[中]返回一个Observable，该Observable发出源Observable发出的项，并转换为指定类型。
调度程序：默认情况下，cast不会在特定调度程序上运行。

代码示例

代码示例来源：origin: ReactiveX/RxNetty

@Override
public Observable<ByteBuf> call(Observable<ByteBuf> upstream) {
  return upstream
    .collect(
      new Func0<CompositeByteBuf>() {
        @Override
        public CompositeByteBuf call() {
          return Unpooled.compositeBuffer();
        }
      },
      new Action2<CompositeByteBuf, ByteBuf>() {
        @Override
        public void call(CompositeByteBuf collector, ByteBuf buf) {
          long newLength = collector.readableBytes() + buf.readableBytes();
          if (newLength <= maxBytes) {
            collector.addComponent(true, buf);
          } else {
            collector.release();
            buf.release();
            throw new TooMuchDataException("More than " + maxBytes + "B received");
          }
        }
      }
    )
    .cast(ByteBuf.class);
}

代码示例来源：origin: PipelineAI/pipeline

@Override
public Observable<Void> mapResponseToRequests(Observable<BatchReturnType> batchResponse, final Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests) {
  return batchResponse.single().doOnNext(new Action1<BatchReturnType>() {
    @Override
    public void call(BatchReturnType batchReturnType) {
      // this is a blocking call in HystrixCollapser
      self.mapResponseToRequests(batchReturnType, requests);
    }
  }).ignoreElements().cast(Void.class);
}

代码示例来源：origin: PipelineAI/pipeline

}).ignoreElements().cast(Void.class);

代码示例来源：origin: davidmoten/rxjava-jdbc

private static <T> Observable<Boolean> commitOrRollbackOnCompleteTransformerIfAtLeastOneValue(
    final boolean isCommit, final Database db, Observable<T> source) {
  CountingAction<T> counter = RxUtil.counter();
  Observable<Boolean> commit = counter
      // get count
      .count()
      // greater than zero or empty
      .filter(greaterThanZero())
      // commit if at least one value
      .compose(db.commitOrRollback_(isCommit));
  return Observable
      // concatenate
      .concat(source
          // count emissions
          .doOnNext(counter)
          // ignore emissions
          .ignoreElements()
          // cast the empty sequence to type Boolean
          .cast(Boolean.class),
          // concat with commit
          commit);
}

代码示例来源：origin: techery/janet

@Override public Observable<ActionState<A>> call(Observable<ActionState> source) {
    return source.cast(type);
  }
}

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

@Override
 public void run() {
  Observable.<Object>just(1, 2, 3).
                   cast(Integer.class).subscribe(new Action1<Integer>() {
   @Override
   public void call(Integer integer) {
    log(integer);
   }
  });
 }
});

代码示例来源：origin: HotelsDotCom/styx

private <T> Observable<T> currentValueIfPresent(String key, Class<T> type) {
  return Observable.from(
      singleOrEmpty(() -> values.get(key)))
      .cast(type);
}

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

@Override
public Observable<Void> mapResponseToRequests(Observable<BatchReturnType> batchResponse, final Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests) {
  return batchResponse.single().doOnNext(new Action1<BatchReturnType>() {
    @Override
    public void call(BatchReturnType batchReturnType) {
      // this is a blocking call in HystrixCollapser
      self.mapResponseToRequests(batchReturnType, requests);
    }
  }).ignoreElements().cast(Void.class);
}

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

@Test
public void testSendByte() {
  AssertableSubscriber<String> subscriber =
      subscribe(destination, byte[].class).map(Message::getPayload).cast(byte[].class).map(String::new)
                        .test();
  sendSubscribeOperation.sendRx(destination, byteMessage);
  subscriber.assertValue(payload).assertNoErrors();
}

代码示例来源：origin: Microsoft/spring-cloud-azure

@Test
public void testSendByte() {
  AssertableSubscriber<String> subscriber =
      subscribe(destination, byte[].class).map(Message::getPayload).cast(byte[].class).map(String::new)
                        .test();
  sendSubscribeOperation.sendRx(destination, byteMessage);
  subscriber.assertValue(payload).assertNoErrors();
}

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

@Test
public void testSendUser() {
  AssertableSubscriber<User> subscriber =
      subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  sendSubscribeOperation.sendRx(destination, userMessage);
  subscriber.assertValue(user).assertNoErrors();
}

代码示例来源：origin: Microsoft/spring-cloud-azure

@Test
public void testSendString() {
  AssertableSubscriber<String> subscriber =
      subscribe(destination, String.class).map(Message::getPayload).cast(String.class).test();
  sendSubscribeOperation.sendRx(destination, stringMessage);
  subscriber.assertValue(payload).assertNoErrors();
}

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

@Test
public void testSendString() {
  AssertableSubscriber<String> subscriber =
      subscribe(destination, String.class).map(Message::getPayload).cast(String.class).test();
  sendSubscribeOperation.sendRx(destination, stringMessage);
  subscriber.assertValue(payload).assertNoErrors();
}

代码示例来源：origin: Microsoft/spring-cloud-azure

@Test
public void testSendUser() {
  AssertableSubscriber<User> subscriber =
      subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  sendSubscribeOperation.sendRx(destination, userMessage);
  subscriber.assertValue(user).assertNoErrors();
}

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

void connect() {
  ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
  Observable<String> txtMessages = observe(connectionFactory, new ActiveMQTopic("orders"))
      .cast(TextMessage.class)
      .flatMap(m -> {
        try {
          return Observable.just(m.getText());
        } catch (JMSException e) {
          return Observable.error(e);
        }
      });
}

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

@Test
public void sample_35() throws Exception {
  CouchbaseAsyncCluster cluster = CouchbaseAsyncCluster.create();
  cluster
      .openBucket("travel-sample")
      .flatMap(cl -> cl.get("route_14197")
          .map(AbstractDocument::content)
          .map(jsonObject -> jsonObject.getArray("schedule")))
      .concatMapIterable(JsonArray::toList)
      .cast(Map.class)
      .filter(m -> ((Number) m.get("day")).intValue() == 0)
      .map(m -> m.get("flight").toString())
      .subscribe(flight -> System.out.println(flight));
}

代码示例来源：origin: Microsoft/spring-cloud-azure

@Test
public void testSendReceiveWithRecordCheckpointMode() {
  setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.RECORD).build());
  AssertableSubscriber<User> subscriber =
      subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  Arrays.stream(messages).forEach(m -> sendSubscribeOperation.sendRx(destination, m));
  subscriber.assertValueCount(messages.length).assertNoErrors();
  verifyCheckpointSuccessCalled(messages.length);
}

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

@Test
public void testSendReceiveWithRecordCheckpointMode() {
  setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.RECORD).build());
  AssertableSubscriber<User> subscriber =
      subscribe(destination, User.class).map(Message::getPayload).cast(User.class).test();
  Arrays.stream(messages).forEach(m -> sendSubscribeOperation.sendRx(destination, m));
  subscriber.assertValueCount(messages.length).assertNoErrors();
  verifyCheckpointSuccessCalled(messages.length);
}

代码示例来源：origin: Microsoft/spring-cloud-azure

@Test
public void testSendReceiveWithManualCheckpointMode() {
  setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.MANUAL).build());
  Observable<Message<?>> observable = subscribe(destination, User.class);
  AssertableSubscriber<User> userSubscriber = observable.map(Message::getPayload).cast(User.class).test();
  sendSubscribeOperation.sendRx(destination, userMessage);
  userSubscriber.assertValue(user).assertNoErrors();
  verifyCheckpointSuccessCalled(0);
}

代码示例来源：origin: com.microsoft.azure/spring-integration-azure-test

@Test
public void testSendReceiveWithManualCheckpointMode() {
  setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.MANUAL).build());
  Observable<Message<?>> observable = subscribe(destination, User.class);
  AssertableSubscriber<User> userSubscriber = observable.map(Message::getPayload).cast(User.class).test();
  sendSubscribeOperation.sendRx(destination, userMessage);
  userSubscriber.assertValue(user).assertNoErrors();
  verifyCheckpointSuccessCalled(0);
}

内容来源于网络，如有侵权，请联系作者删除！

Observable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Observable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com