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

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

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

Observable.subscribe介绍

[英]Subscribes to an Observable but ignore its emissions and notifications. Scheduler: subscribe does not operate by default on a particular Scheduler.
[中]订阅可观察到的，但忽略其发射和通知。调度程序：订阅默认情况下不会在特定调度程序上运行。

代码示例

代码示例来源：origin: greenrobot/greenDAO

  1. private void updateNotes() {
  2.   notesQuery.list()
  3.       .observeOn(AndroidSchedulers.mainThread())
  4.       .subscribe(new Action1<List<Note>>() {
  5.         @Override
  6.         public void call(List<Note> notes) {
  7.           notesAdapter.setNotes(notes);
  8.         }
  9.       });
  10. }

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

  1. public void startCachingStreamValuesIfUnstarted() {
  2.   if (rollingMaxSubscription.get() == null) {
  3.     //the stream is not yet started
  4.     Subscription candidateSubscription = observe().subscribe(rollingMax);
  5.     if (rollingMaxSubscription.compareAndSet(null, candidateSubscription)) {
  6.       //won the race to set the subscription
  7.     } else {
  8.       //lost the race to set the subscription, so we need to cancel this one
  9.       candidateSubscription.unsubscribe();
  10.     }
  11.   }
  12. }

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

  1. public void startCachingStreamValuesIfUnstarted() {
  2.   if (rollingDistributionSubscription.get() == null) {
  3.     //the stream is not yet started
  4.     Subscription candidateSubscription = observe().subscribe(rollingDistribution);
  5.     if (rollingDistributionSubscription.compareAndSet(null, candidateSubscription)) {
  6.       //won the race to set the subscription
  7.     } else {
  8.       //lost the race to set the subscription, so we need to cancel this one
  9.       candidateSubscription.unsubscribe();
  10.     }
  11.   }
  12. }

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

  1. public void startCachingStreamValuesIfUnstarted() {
  2.   if (subscription.get() == null) {
  3.     //the stream is not yet started
  4.     Subscription candidateSubscription = observe().subscribe(counterSubject);
  5.     if (subscription.compareAndSet(null, candidateSubscription)) {
  6.       //won the race to set the subscription
  7.     } else {
  8.       //lost the race to set the subscription, so we need to cancel this one
  9.       candidateSubscription.unsubscribe();
  10.     }
  11.   }
  12. }

代码示例来源：origin: greenrobot/greenDAO

  1. @Override
  2.   public void onNoteClick(int position) {
  3.     Note note = notesAdapter.getNote(position);
  4.     final Long noteId = note.getId();
  5.     noteDao.deleteByKey(noteId)
  6.         .observeOn(AndroidSchedulers.mainThread())
  7.         .subscribe(new Action1<Void>() {
  8.           @Override
  9.           public void call(Void aVoid) {
  10.             Log.d("DaoExample", "Deleted note, ID: " + noteId);
  11.             updateNotes();
  12.           }
  13.         });
  14.   }
  15. };

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

  1. @Test
  2. public void noEvents() throws InterruptedException {
  3.   CountDownLatch commandLatch = new CountDownLatch(1);
  4.   CountDownLatch threadPoolLatch = new CountDownLatch(1);
  5.   Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch);
  6.   readCommandStream.observe().take(1).subscribe(commandSubscriber);
  7.   Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch);
  8.   readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber);
  9.   //no writes
  10.   assertFalse(commandLatch.await(1000, TimeUnit.MILLISECONDS));
  11.   assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS));
  12. }

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

  1. @Test
  2. public void noEvents() throws InterruptedException {
  3.   CountDownLatch latch = new CountDownLatch(1);
  4.   Subscriber<HystrixCommandCompletion> subscriber = getLatchedSubscriber(latch);
  5.   commandStream.observe().take(1).subscribe(subscriber);
  6.   //no writes
  7.   assertFalse(latch.await(1000, TimeUnit.MILLISECONDS));
  8. }

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

  1. @Test
  2. public void testSemaphoreIsolatedSuccess() throws Exception {
  3.   CountDownLatch commandLatch = new CountDownLatch(1);
  4.   CountDownLatch threadPoolLatch = new CountDownLatch(1);
  5.   Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch);
  6.   readCommandStream.observe().take(1).subscribe(commandSubscriber);
  7.   Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch);
  8.   readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber);
  9.   ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS);
  10.   writeToStream.executionDone(result, commandKey, threadPoolKey);
  11.   assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS));
  12.   assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS));
  13. }

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

  1. @Test
  2. public void testSemaphoreIsolatedTimeout() throws Exception {
  3.   CountDownLatch commandLatch = new CountDownLatch(1);
  4.   CountDownLatch threadPoolLatch = new CountDownLatch(1);
  5.   Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch);
  6.   readCommandStream.observe().take(1).subscribe(commandSubscriber);
  7.   Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch);
  8.   readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber);
  9.   ExecutionResult result = ExecutionResult.from(HystrixEventType.TIMEOUT);
  10.   writeToStream.executionDone(result, commandKey, threadPoolKey);
  11.   assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS));
  12.   assertFalse(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS));
  13. }

代码示例来源：origin: greenrobot/greenDAO

  1. static <T> TestSubscriber<T> awaitTestSubscriber(Observable<T> observable) {
  2.   TestSubscriber<T> testSubscriber = new TestSubscriber<>();
  3.   observable.subscribe(testSubscriber);
  4.   testSubscriber.awaitTerminalEvent(3, TimeUnit.SECONDS);
  5.   testSubscriber.assertNoErrors();
  6.   testSubscriber.assertCompleted();
  7.   return testSubscriber;
  8. }

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

  1. @Test
  2. public void testThreadIsolatedSuccess() throws InterruptedException {
  3.   CountDownLatch commandLatch = new CountDownLatch(1);
  4.   CountDownLatch threadPoolLatch = new CountDownLatch(1);
  5.   Subscriber<HystrixCommandCompletion> commandSubscriber = getLatchedSubscriber(commandLatch);
  6.   readCommandStream.observe().take(1).subscribe(commandSubscriber);
  7.   Subscriber<HystrixCommandCompletion> threadPoolSubscriber = getLatchedSubscriber(threadPoolLatch);
  8.   readThreadPoolStream.observe().take(1).subscribe(threadPoolSubscriber);
  9.   ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS).setExecutedInThread();
  10.   writeToStream.executionDone(result, commandKey, threadPoolKey);
  11.   assertTrue(commandLatch.await(1000, TimeUnit.MILLISECONDS));
  12.   assertTrue(threadPoolLatch.await(1000, TimeUnit.MILLISECONDS));
  13. }

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

  1. @Test
  2. public void testObservableRaiseHystrixRuntimeException() {
  3.   TestSubscriber<Void> testSubscriber = new TestSubscriber<Void>();
  4.   service.observableCommandShouldRaiseHystrixRuntimeException().subscribe(testSubscriber);
  5.   testSubscriber.assertError(HystrixRuntimeException.class);
  6. }

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

  1. @Test
  2. public void testPropagateCauseException() throws NotFoundException {
  3.   TestSubscriber<Void> testSubscriber = new TestSubscriber<Void>();
  4.   userService.deleteUser("").subscribe(testSubscriber);
  5.   testSubscriber.assertError(NotFoundException.class);
  6. }

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

  1. @Test
  2. public void testDoNotInterruptObserveOnTimeoutIfPropertySaysNotTo() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), false);
  5.   // when
  6.   cmd.observe().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertFalse(cmd.hasBeenInterrupted());
  10. }

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

  1. @Test
  2. public void testInterruptToObservableOnTimeout() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true);
  5.   // when
  6.   cmd.toObservable().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertTrue(cmd.hasBeenInterrupted());
  10. }

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

  1. @Test
  2. public void testSingleWriteSingleSubscriber() throws InterruptedException {
  3.   CountDownLatch latch = new CountDownLatch(1);
  4.   Subscriber<HystrixCommandCompletion> subscriber = getLatchedSubscriber(latch);
  5.   commandStream.observe().take(1).subscribe(subscriber);
  6.   ExecutionResult result = ExecutionResult.from(HystrixEventType.SUCCESS).setExecutedInThread();
  7.   HystrixCommandCompletion event = HystrixCommandCompletion.from(result, commandKey, threadPoolKey);
  8.   commandStream.write(event);
  9.   assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
  10. }

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

  1. @Test
  2. public void testInterruptObserveOnTimeout() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true);
  5.   // when
  6.   cmd.observe().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertTrue(cmd.hasBeenInterrupted());
  10. }

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

  1. @Test
  2. public void testInterruptObserveOnTimeout() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true);
  5.   // when
  6.   cmd.observe().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertTrue(cmd.hasBeenInterrupted());
  10. }

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

  1. @Test
  2. public void testDoNotInterruptToObservableOnTimeoutIfPropertySaysNotTo() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), false);
  5.   // when
  6.   cmd.toObservable().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertFalse(cmd.hasBeenInterrupted());
  10. }

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

  1. @Test
  2. public void testInterruptToObservableOnTimeout() throws InterruptedException {
  3.   // given
  4.   InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true);
  5.   // when
  6.   cmd.toObservable().subscribe();
  7.   // then
  8.   Thread.sleep(500);
  9.   assertTrue(cmd.hasBeenInterrupted());
  10. }

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

Observable

相关文章

    查看更多

    热门标签

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

    最新文章

    更多

    Observable类方法

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