Wie aus rxJava extrahieren und auf Realm Offline-Datenbank setzen?

Question

Ich muss eine lange Liste von 30k Flughäfen herunterladen und auf eine Offline-Datenbank stellen.

Ich habe diesen Code, um die json aus dem Internet heruntergeladen werden:

bFetch.setOnClickListener(new View.OnClickListener() { 
     public void onClick(View v) 
     { 
      Retrofit retrofit = new Retrofit.Builder() 
        .addConverterFactory(GsonConverterFactory.create()) 
        .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) 
        .baseUrl(GithubService.SERVICE_ENDPOINT).build(); 

      GithubService service = retrofit.create(GithubService.class); 

      service.getAirport() 
        .subscribeOn(Schedulers.newThread()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Subscriber<List<Airport>>() { 
         @Override 
         public void onCompleted() 
         { 
         bClear.setText("OK"); 
         } 

         @Override 
         public void onError(Throwable e) { 

         } 

         @Override 
         public void onNext(List<Airport> airports) 
         { 
         Log.d("msh",String.valueOf(airports.size())); 

         } 
        }); 

     } 
    }); 

und es funktioniert sehr gut, aber wenn ich nur ein Objekt extrahieren mag, wie Karte oder einen flatMap, es gibt mir diese:

service.getAirport() 
        .map(new Func1<List<Airport>, Airport>() 
        { 
         @Override 
         public Airport call(List<Airport> airports) { 
          return null; 
         } 
        }) 
        .subscribeOn(Schedulers.newThread()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Subscriber<List<Airport>>() { 
         @Override 
         public void onCompleted() 
         { 
         bClear.setText("OK"); 
         } 

         @Override 
         public void onError(Throwable e) { 

         } 

         @Override 
         public void onNext(List<Airport> airports) 
         { 
         Log.d("msh",String.valueOf(airports.size())); 

         } 
        }); 

     } 
    }); 

mit dem Fehler:

Cannot resolve method 'subscribe(anonymous rx.Subscriber>)

so:

1. was muss ich tun, um es zu lösen? Mein Problem ist, dass ich nicht sehr gut rX verstehe und ich habe auch ein wenig Verwirrung
2. könnte ich Daten in Realm-Datenbank in map() -Methode (wenn es funktioniert)?

Danke

Answer 1

Da man von einem List<Airport> Mapping ist zu einem Airport Sie ein Subscriber<Airport> statt Subscriber<List<Airport>>, zusammen mit der gleichen Änderung des onNext Verfahren haben müssen.

Answer 2

sieht aus wie es mit Java8 und RxJava2-RC5 kompilieren würde. Ich änderte Teilnehmer param von Liste zu X und die onNext Methode von Liste zu X. Vielleicht könnten Sie etwas mehr Informationen über Ihre Umwelt zur Verfügung stellen. Bitte beachten Sie, dass das Zurückgeben von NULL in RxJava2 nicht mehr möglich ist.

Beachten Sie außerdem, dass die Verwendung von NewThread-Scheduler keine gute Idee ist.

This scheduler simply starts a new thread every time it is requested via subscri beOn() or observeOn() . newThread() is hardly ever a good choice, not only because of the latency involved when starting a thread, but also because this thread is not reused. --Tomasz Nurkiewicz from "Reactive Programming with RxJava"

Beispiel-Impl mit RxJava2-RC5

Observable.just(Arrays.asList("1", "2", "3")) 
       .map(new Function<List<String>, String>() { 
        @Override 
        public String apply(List<String> s) throws Exception { 
         return null; 
        } 
       }).subscribeOn(Schedulers.newThread()) 
       .subscribe(new Observer<String>() { 
        @Override 
        public void onSubscribe(Disposable d) { 

        } 

        @Override 
        public void onNext(String value) { 

        } 

        @Override 
        public void onError(Throwable e) { 

        } 

        @Override 
        public void onComplete() { 

        } 
       });