分块流不符合预期的Retrofit + RxJava

Question

@GET("poll/session/{sessionId}/details")
Observable getSessionDetails(@Path("sessionId") String sessionId);

@GET("poll/session/{sessionId}/details")
@Streaming
Observable getSessionDetails(@Path("sessionId") String sessionId);

@Override
public Observable getSessionDetails(String sessionId) {
return sessionAPI.getSessionDetails(sessionId)
.flatMap(responseBody -> events(responseBody.source()));
}

public static Observable<String> events(BufferedSource source) {
    return Observable.create(subscriber -> {
        try {
            while (!source.exhausted()) {
                subscriber.onNext(source.readUtf8Line());
            }
        } catch (IOException e) {
            e.printStackTrace();
            subscriber.onError(e);
        }
        subscriber.onCompleted();
    });
}

除非完成所有块,否则不会调用events()方法.

但是,大块的流预计会被大块交付,这似乎并没有发生.

我已尝试使用和不使用@Streaming注释到API,但行为是相同的.

我曾经使用Android Retrofit 2 + RxJava:听取无休止的流作为参考来做我的实现

Answer 1

好吧,我找到了答案.这是因为我正在使用Body属性进行记录

logging.setLevel(HttpLoggingInterceptor.Level.BODY);

因此,由于记录器正在等待整个主体打印它,它的行为与问题中提到的方式相同.

参考:Square的Retrofit响应解析逻辑:流媒体？