RxJava - Наблюдаемые: один наблюдаемый, чем возврат другой Наблюдаемый

Question

Я работаю в приложении, чем конвертирую один список объектов в другие объекты, для этого я использую Observable of Rx Java.

У меня есть два метода: один из них - это адаптер (фасад Джерси), у которого есть служба асин, и этот метод потребляет от другой службы, наблюдаемой.

Мне нужно потреблять Наблюдаемое и обрабатывать каждый элемент, когда это завершено. Я создал некий список всех обработанных элементов и возвращаю новый Observable, где E - список обрабатываемых элементов.

Для обработки каждого элемента я использую оператор flatMap, но я не знаю, как создать новый Observer, у которого есть один тип diferente, например, List of all, обработанный оператором flatmap.

Любая идея?

Благодаря

Update:

Этот код, обработать каждый элемент и есть возвращать другой Observable, но я не знаю, если это хорошо сделано.

@Override 
public Observable<ArrayList> getGeoJson2() { 
    WKTReader2 reader2 = new WKTReader2(); 
    WKBReader wkbReader = new WKBReader(); 
    ArrayList featureCollection = new ArrayList(); 

    Subject<ArrayList,ArrayList> subject = PublishSubject.create(); 

    manzanaRepository.getManzanas().map(new Func1<Manzana, SimpleFeature>() { 
     @Override 
     public SimpleFeature call(Manzana manzana) { 
      try { 
       SimpleFeatureType TYPE = DataUtilities.createType("", "geom,name:String"); 
       return SimpleFeatureBuilder.build(TYPE, new Object[]{ null, "name1"}, null); 
      }catch (Exception e){ 
       System.out.println(e.getMessage()); 
       return null; 
      } 

     } 
    }).subscribe(new Subscriber<SimpleFeature>() { 
     @Override 
     public void onCompleted() { 
      subject.onNext(featureCollection); 
      subject.onCompleted(); 
     } 

     @Override 
     public void onError(Throwable throwable) { 
      subject.onError(throwable); 
     } 

     @Override 
     public void onNext(SimpleFeature simpleFeature) { 
      featureCollection.add(simpleFeature); 
     } 
    }); 

    return subject; 
} 

И этот код является тот, кто использует Observable возвратившихся:

@GET 
@Produces(MediaType.APPLICATION_JSON) 
@Path("/async/geom") 
public void asyncGetGeom(@Suspended final AsyncResponse asyncResponse) { 

    Observable<ArrayList> features = service.getGeoJson2(); 

    features.subscribe(new Observer<ArrayList>() { 
     @Override 
     public void onCompleted() { 

      System.out.println("Se completo la accion!!!"); 
     } 

     @Override 
     public void onError(Throwable throwable) { 

      System.out.println(throwable.getMessage()); 
     } 

     @Override 
     public void onNext(ArrayList features) { 
      asyncResponse.resume(features); 
     } 
    }); 
} 

Метод onNext() никогда не вызывается !!!

Благодаря

Answer 1

Попробуйте заменить getGeoJson2() с этим:

@Override 
public Observable<List<SimpleFeature>> getGeoJson2() { 
    return manzanaRepository.getManzanas() 
      .map(new Func1<Manzana, SimpleFeature>() { 
       @Override 
       public SimpleFeature call(Manzana manzana) { 
        try { 
         SimpleFeatureType TYPE = DataUtilities.createType("", "geom,name:String"); 
         return SimpleFeatureBuilder.build(TYPE, new Object[]{ null, "name1"}, null)); 
        } catch (Exception e) { 
         System.out.println(e.getMessage()); 
         return null; 
        } 
       } 
      }) 
      .filter(new Func1<SimpleFeature, Boolean>() { 
       @Override 
       public Boolean call(SimpleFeature sf) { 
        return sf != null; 
       } 
      }) 
      .toList(); 
} 

Объяснение:toList() используется оператор, который ждет onCompleted от источника наблюдаемого и излучает все элементы, излучаемые источником наблюдаемый как список.

Answer 2

проблема с вашим кодом является то, что вы используете PublishSubject, который завершает прежде чем начать использовать его

, если вы измените его на BehaviorSubject, он будет работать

однако, лучше (и гораздо короче) для реализации это следующим образом:

manzanaRepository.getManzanas().map(/**/).collect(/*create array*/, /*add next item*/)