使用 Retrofit 来消费服务器发送的事件答案

【问题标题】：Use Retrofit to consume Server Sent Events使用 Retrofit 来消费服务器发送的事件
【发布时间】：2015-11-14 14:04:44
【问题描述】：

我正在尝试使用将服务器发送的事件发送到客户端的 rest api [1]。我目前正在使用方形改造来使用它，但我不知道该怎么做。有过改造经验的人可以帮忙吗？如果不进行改造，请推荐其他可以做到这一点的 Java 库。

[1]https://mesosphere.github.io/marathon/docs/rest-api.html#get-v2-events

【问题讨论】：

标签： java rest retrofit

【解决方案1】：

试试这个库：oksee。

OkSse 是 OkHttp 的扩展库，用于创建服务器发送事件 (SSE) 客户端

由于我遇到了同样的问题，根据我的研究，这是目前最好的选择，因为 Retrofit 不支持它。

https://github.com/heremaps/oksse

【讨论】：

PProject 似乎死了。现在有：github.com/launchdarkly/okhttp-eventsource

【解决方案2】：

没有必要将 Retrofit 和 SSE 混为一谈。使用改造来获取输入流，然后找到（或编写）一个输入流解析器来分块 SSE 事件。

在改造中我有这个：

public interface NotificationAPI {
    @GET("notifications/sse")
    Call<InputStream> getNotificationsStream(@retrofit2.http.Query("Last-Event-ID") String lastEventId);
}

我为InputStream写了一个快速转换器工厂：

public class InputStreamConverterFactory extends Converter.Factory {

    private static class InputStreamConverter implements Converter<ResponseBody, InputStream> {
        @Nullable
        @Override
        public InputStream convert(ResponseBody value) throws IOException {
            return value.byteStream();
        }
    }

    @Override
    public @Nullable
    Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations, Retrofit retrofit) {
        if (type.equals(InputStream.class)) {
            return new InputStreamConverter();
        }
        return null;
    }
}

我的客户端代码如下所示：

var cinputStream = api.getNotificationsStream(null);
var inputStream = cinputStream.execute().body();
try(var sseStream = new SSEStreamParser(inputStream)) {
   //handle the stream here...
}

有一个 OkHttp SSE 解析器，您可能可以使用它。然而：

OkHttp SSE 代码带有线程。您可能想要使用自己的线程模型。
实际的 OkHttp SSE 解析器是一个内部包。这并不适合举重。

【讨论】：

【解决方案3】：

我知道这是个老问题。但是我没有找到完整的示例，现在尝试通过我的代码提供它。 我们只使用retrofit 和coroutines

1.在retrofit API interface需要添加代码。注意我们使用@Streaming和返回类型Call<ResponseBody>

@POST("/v1/calc/group-prices")
@Streaming
fun calculateGroupPrices(@Body listOptions: List<GroupCalculatorOptions>): Call<ResponseBody>

2.在您的repository 类中需要添加此代码。注意我们使用的flow 并阅读stream。要了解带有有效负载的消息已经到达，它必须以"data:"开头

fun loadGroupDeliveryRateInfos(listOptions: List<GroupCalculatorOptions>) = flow {
        coroutineScope {
            val response = restApi.calculateGroupPrices(listOptions).execute()
            if (response.isSuccessful) {
                val input = response.body()?.byteStream()?.bufferedReader() ?: throw Exception()
                try {
                    while (isActive) {
                        val line = input.readLine() ?: continue
                        if (line.startsWith("data:")) {
                            try {
                                val groupDeliveryRateInfo = gson.fromJson(
                                    line.substring(5).trim(),
                                    GroupDeliveryRateInfo::class.java
                                )
                                emit(groupDeliveryRateInfo)
                            } catch (e: Exception) {
                                e.printStackTrace()
                            }
                        }
                    }
                } catch (e: IOException) {
                    throw Exception(e)
                } finally {
                    input.close()
                }
            } else {
                throw HttpException(response)
            }
        }
    }

3.最后一步，我们需要在ViewModel 中收集我们的数据。我们只需要调用来自repository的方法

 repository.loadGroupDeliveryRateInfos(it.values.toList())
                .collect { info ->
                    handleGroupDeliveryRateInfo(info)
                }

仅此而已，不需要额外的库。

【讨论】：