Почему это наблюдаемое испускает только одно значение

Question

У меня есть следующий код, основанный на примере, представленном @ a.bertucci здесь Emit objects for drawing in the UI in a regular interval using RxJava on Android, где я фиксирую наблюдаемый с помощью таймера. Когда я вызываю подписку, вызывая processDelayedItems(), код [A] в zipped Observable выполняется ровно один раз, и один элемент испускается на [B]. Я бы ожидал, что код [A] будет запускаться непрерывно после запуска и продолжит излучать элементы каждые 1500 мс, но он, очевидно, работает только один раз здесь.

private static void processDelayedItems() { 

    Observable.zip(
      Observable.create(new Observable.OnSubscribe<Object>() { 

       @Override public void call(Subscriber<? super Object> subscriber) { 
        // [A] this code is only called once 
        subscriber.OnNext(o) 
       } 

      }), 
      Observable.timer(1500, 1500, TimeUnit.MILLISECONDS), new Func2<Object, Long, Object>() { 
       @Override public Object call(Object entity, Long aLong) { 
        return entity; 
       } 
      } 
    ) 
    .subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<Object>() { 

       @Override public void call(Object entity) { 
        // ... and accordingly one item is emitted [B] 
       } 

      }, new Action1<Throwable>() { 

       @Override public void call(Throwable throwable) { 
        throwable.printStackTrace(); 
       } 

      }, new Action0() { 

       @Override public void call() { 

       } 

      }); 

} 

1. Может кто-нибудь увидеть проблему, которую я здесь? Нужно ли мне ссылаться на Observable из-за пределов функции, чтобы поддерживать ее в течение более продолжительного времени? Собирается ли он GC (Android)? Проблема в том, что эта функция является статической?

2. Каковы правила наблюдений с точки зрения их жизни? Есть ли какие-либо рекомендации о том, как следует использовать ссылки на Observables, и если они могут быть статическими вообще? В моих тестах я заметил, что это не имеет большого значения, но, возможно, здесь, когда задействован таймер.

-

Исправленный код [не работает еще]:

• добавил повтор()

  Observable.zip(
       Observable.create(new Observable.OnSubscribe<Object>() { 
  
        @Override public void call(Subscriber<? super Object> subscriber) { 
         // [A] this code is only called once 
         subscriber.OnNext(o); 
         subscriber.OnCompleted(); 
        } 
  
       }).repeat(Schedulers.newThread()), 
       Observable.timer(1500, 1500, TimeUnit.MILLISECONDS), new Func2<Object, Long, Object>() { 
        @Override public Object call(Object entity, Long aLong) { 
         return entity; 
        } 
       } 
  ) 
  .subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(new Action1<Object>() { 
  
        @Override public void call(Object entity) { 
         // ... and accordingly one item is emitted [B] 
        } 
  
       }, new Action1<Throwable>() { 
  
        @Override public void call(Throwable throwable) { 
         throwable.printStackTrace(); 
        } 
  
       }, new Action0() { 
  
        @Override public void call() { 
  
        } 
  
       }); 
  

Answer 1

Вы должны repeat генерировать бесконечное Observable , НАПРИМЕР,

Observable.create(new Observable.OnSubscribe<Object>() { 

     @Override public void call(Subscriber<? super Object> subscriber) { 
      // [A] this code is only called once 
      if (!subscriber.isUnsubscribed()) { 
       subscriber.onNext(o); 
      } 
      if (!subscriber.isUnsubscribed()) { 
       subscriber.onCompleted(); 
      } 
     } 

    }).repeat(Schedulers.newThread()); 

ли, что мне нужно ссылаться на Наблюдаемые извне функции, чтобы сохранить его в живых больше времени? Собирается ли он GC (Android)? Проблема в том, что эта функция является статической?

Поскольку вы используете Schedulers.newThread() и timer, там будут какие-то другие темы, которые есть ссылка на ваш Observable. Вам не нужно больше работать.

Каковы правила наблюдений с точки зрения их жизни? Есть ли какие-либо рекомендации о том, как следует использовать ссылки на Observables, и если они могут быть статическими вообще? В моих тестах я заметил, что это не имеет большого значения, но, возможно, здесь, когда задействован таймер.

Вы правы. Это не имеет значения.

Answer 2

В связи с Вашим комментарием, для простоты, вы могли бы сделать это,

Observable.timer(1500, 1500, TimeUnit.MILLISECONDS) 
     .flatMap(new Func1<Long, Observable<Object>>() { 
      @Override 
      public Observable<Object> call(Long aLong) { 
       String o = "0"; 
       return Observable.from(o); 
      } 
     }) 
     .subscribe(new Action1<Object>() { 
      @Override 
      public void call(Object aLong) { 
       System.out.println(aLong); 
      } 
     }); 

Здесь вы все еще получаете преимущество таймера без добавленного почтового индекса/повтора на вершине. Это все еще немного многословно, но это немного проще.