RxJava + Retrofit polling lungo

Question

Il mio problema è che non riesco a ottenere lo streaming infinito con Retrofit. Dopo aver ottenuto le credenziali per la richiesta di poll() iniziale - faccio la richiesta di polling() iniziale. Ogni richiesta di poll() risponde in 25 secondi se non ci sono cambiamenti, o prima se ci sono cambiamenti - return changed_data []. Ogni risposta contiene i dati timestamp necessari per la successiva richiesta di polling: dovrei fare una nuova richiesta di poll() dopo ogni risposta poll(). Qui è il mio codice:

getServerApi().getLongPollServer() 
    .flatMap(longPollServer -> getLongPollServerApi(longPollServer.getServer()).poll("a_check", Config.LONG_POLLING_SERVER_TIMEOUT, 2, longPollServer.getKey(), longPollServer.getTs(), "") 
    .take(1) 
    .flatMap(longPollEnvelope -> getLongPollServerApi(longPollServer.getServer()).poll("a_check", Config.LONG_POLLING_SERVER_TIMEOUT, 2, longPollServer.getKey(), longPollEnvelope.getTs(), ""))) 
    .retry() 
    .subscribe(longPollEnvelope1 -> { 
    processUpdates(longPollEnvelope1.getUpdates()); 
}); 

Sono nuovo di RxJava, forse non capisco qualcosa, ma non riesco a ottenere flusso infinito. Ottengo 3 chiamate, quindi onNext e onComplete.

P.S. Forse c'è una soluzione migliore per implementare il polling lungo su Android?

Answer 1

Sebbene non sia l'ideale, credo che potresti utilizzare gli effetti collaterali di RX per ottenere un risultato desiderato (operazioni "doOn").

Observable<CredentialsWithTimestamp> credentialsProvider = Observable.just(new CredentialsWithTimestamp("credentials", 1434873025320L)); // replace with your implementation 

Observable<ServerResponse> o = credentialsProvider.flatMap(credentialsWithTimestamp -> { 
    // side effect variable 
    AtomicLong timestamp = new AtomicLong(credentialsWithTimestamp.timestamp); // computational steering (inc. initial value) 
    return Observable.just(credentialsWithTimestamp.credentials) // same credentials are reused for each request - if invalid/onError, the later retry() will be called for new credentials 
      .flatMap(credentials -> api.query("request", credentials, timestamp.get())) // this will use the value from previous doOnNext 
      .doOnNext(serverResponse -> timestamp.set(serverResponse.getTimestamp())) 
      .repeat(); 
}) 
     .retry() 
     .share(); 

private static class CredentialsWithTimestamp { 

    public final String credentials; 
    public final long timestamp; // I assume this is necessary for you from the first request 

    public CredentialsWithTimestamp(String credentials, long timestamp) { 
     this.credentials = credentials; 
     this.timestamp = timestamp; 
    } 
} 

Quando si sottoscrive a "o", l'osservabile interno si ripeterà. Se si verifica un errore, "o" riproverà e richiederà nuovamente il flusso di credenziali.

Nell'esempio, lo sterzo computazionale viene ottenuto aggiornando la variabile timestamp, che è necessaria per la richiesta successiva.