Подписываться и отменить подписку на каждый выпущенный товар

Question

Я использую reactive-location lib.

Моим вариантом использования является то, что у меня есть поток объектов, излучаемых наблюдаемым. Эти предметы могут выделяться потенциально каждые несколько часов. Как только элемент испускается, я хочу получить местоположение и использовать zipWith (насколько я понимаю) испускать объект, содержащий местоположение.

Проблема заключается в том, что, поскольку объекты будут излучаться только один раз каждые несколько часов, я не могу держать место наблюдаемым горячим, так как оно разряжает батарею.

Поэтому мне нужно следующее: после того, как объект передан в поток, подпишитесь на место, которое можно наблюдать после получения местоположения, отмените подписку на наблюдаемое место. Это необходимо делать постоянно.

Насколько я понимаю, этот трансформатор заботится о отпиской

public <T> Observable.Transformer<T, T> takeNextAndUnsubscribe() { 
    return new Observable.Transformer<T, T>() { 

     @Override 
     public Observable<T> call(Observable<T> tObservable) { 
      final BehaviorSubject subject = BehaviorSubject.create(); 
      Observable source = tObservable.doOnNext(new Action1<T>() { 
       @Override 
       public void call(T t) { 
        subject.onNext(t); 
       } 
      }); 
      return Observable 
        .merge(source.takeUntil(subject), subject) 
        .take(1); 
     } 

    }; 
} 

Но как бы я подписаться снова, как только новый объект отправляется вниз по течению?

Answer 1

Похоже, что вам нужно комбинировать исходные элементы с текущим местоположением при их испускании. Здесь нет ничего нового. Просто используйте flatMap() на каждом из исходных элементов, чтобы объединить его с местоположением.

source.flatMap(item -> 
     locationProvider 
       .getLastKnownLocation() 
       .map(location -> new ItemWithLocation<>(item, location)) 
); 

class ItemWithLocation<T> { 
    private final T item; 
    private final Location location; 

    public ItemWithLocation(T item, Location location) { 
     this.item = item; 
     this.location = location; 
    } 

    public T getItem() { 
     return item; 
    } 

    public Location getLocation() { 
     return location; 
    } 
} 

EDIT: Обновлен вторым примером. Ниже будут подписаны обновления местоположения до тех пор, пока не будет достигнута определенная точность, а затем объедините их с исходным кодом. Ключевым моментом здесь является использование first(). Использование этого будет отписываться от поставщика местоположения всякий раз, когда вы получаете место, которое удовлетворяет вашим потребностям.

LocationRequest request = 
     LocationRequest 
      .create() 
      .setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY) 
      .setInterval(100); 

source.flatMap(item -> 
     locationProvider 
       .getUpdatedLocation(request) 
       .first(location -> location.getAccuracy() < 5.0f) 
       .map(location -> new ItemWithLocation<>(item, location)) 
);