java - RxJava. Observable.delay work strange (lacks some items at the end) -


i'm trying understand rxjava. test code is:

import rx.observable; import rx.subscriber; import rx.functions.action1;  import java.util.concurrent.timeunit;  public class hello {     public static void main(string[] args) {              observable<string> observable = observable.create(new observable.onsubscribe<string>() {             @override             public void call(subscriber<? super string> subscriber) {                 try {                     thread.sleep(1000);                     subscriber.onnext("a");                     thread.sleep(1000);                     subscriber.onnext("b");                     thread.sleep(1000);                     subscriber.onnext("c");                     thread.sleep(1000);                     subscriber.onnext("d");                     thread.sleep(1000);                     subscriber.onnext("e");                     thread.sleep(1000);                     subscriber.onnext("f");                     thread.sleep(1000);                     subscriber.onnext("g");                     thread.sleep(1000);                     subscriber.onnext("h");                 } catch (interruptedexception e) {                     subscriber.onerror(e);                 }             }         });          observable                 .delay(2, timeunit.seconds)                 .subscribe(new action1<string>() {                     @override                     public void call(string string) {                         system.out.println(string);                     }                 });     } }

without .delay(2, timeunit.seconds) have output: b c d e f g h .delay(2, timeunit.seconds) output lacks "g" , "h": b c d e f

how can be? documentation says delay emits items emitted source observable shifted forward in time specified delay

the delay overload using schedules work on different thread , results in implicit race condition.all temporal operators (such delay, buffer, , window) need use scheduler schedule effect later , can result in unexpected race conditions if aren't aware of , use them carefully. in case delay operator schedules work downstream on separate thread pool. here order of execution (on main thread) in test.

  1. your observable subscribed , waits 1000 millis before onnext("a")
  2. next it's received delay. schedules downstream onnext 2 seconds later.
  3. control flow returns observable waits 1000 millis.
  4. observable onnext("b") delay. delay schedules onnext of "b" 2 seconds later.
  5. .... (repeat)
  6. when observable calls onnext("h") schedules work returns subscribe , terminates test (causing scheduled work disappear).

in order execute asynchronously you can schedule delay on trampoline scheduler implementation.

.delay(2, timeunit.seconds, schedulers.trampoline())

Comments

Post a Comment

Popular posts from this blog

javascript - How to bind ViewModel Store to View? -

i'm pretty new ext js , trying embed multiselect inside panel . the viewmodel has stores property can see here: ext.define('test.view.controls.search.searchfiltermodel', { extend: 'ext.app.viewmodel', alias: 'viewmodel.filter', data: { title: '' }, stores: { test: { fields: [ 'id', 'name' ], proxy: { type: 'ajax', url: 'api/test', reader: 'array' }, autoload: true } } }); i bind in view this: viewmodel: { type: 'filter' }, layout: 'fit', border: 1, plain: true, scrollable: 'y', layout: 'fit', bind: { title: '{title}', }, items: { xtype: 'multiselect', scrollable: false, allowblank: true, ddreorder: true, bind: { store: '{test}' }, valuefield: 'id', displayfield: 'name' } ...
Read more

recursion - Can every recursive algorithm be improved with dynamic programming? -

i first year undergraduate csc student looking competitive programming. recursion involves defining , solving sub problems. understand, top down dynamic programming (dp) involves memoizing solutions sub problems reduce time complexity of algorithm. can top down dp used improve efficiency of every recursive algorithm overlapping sub problems? dp fail work , how can identify this? the short answer is: yes. however, there constraints. obvious 1 recursive calls must overlap. i.e. during execution of algorithm, recursive function must called multiple times same parameters. lets truncate recursion tree memoization. can use memoization reduce number of calls. however, reduction of calls comes price. need store results somewhere. next obvious constraint need have enough memory. comes not-so obvious constraint. memory access requires time. first need find result stored , maybe copy location. in cases, might faster let recursion calculate result instead of loading somewher...
Read more

c - Why does alarm() cause fgets() to stop waiting? -

i playing around signals in c. main function asks input using fgets(name, 30, stdin) , , sits there , waits. set alarm alarm(3) , , reassigned sigalrm call function myalarm calls system("say pay attention") . after alarm goes off, fgets() stops waiting input , main fn continues on. happens if change myalarm set variable , nothing it. void myalarm(int sig) { //system("say pay attention"); int x = 0; } int catch_signal(int sig, void (*handler)(int)) { // when signal comes in, "catch" , "handle it" in way want struct sigaction action; // create new sigaction action.sa_handler = handler; // set it's sa_handler attribute function specified in header sigemptyset(&action.sa_mask); // "turn signals in sa_mask off?" "set sa_mask contian no signals, i.e. nothing masked?" action.sa_flags = 0; // not sure, looks aren't using of available flags, whatever may return sigaction(sig, ...
Read more