使用 SSE(Server-Sent Events) 进行 HTTP 服务器推送
这个示例是一个类似 twitter 的 web 应用程序，使用 Server-Sent Events 来支持实时刷新。
运行
docker-compose up
然后, 浏览 http://localhost:8080
您可以添加自己的帖子或点击按钮获得随机生成的帖子。
无论哪种方式，feeds 列表和 feed 中的帖子都应该是最新的。尝试使用第二个浏览器窗口查看更新。
它是如何工作的
可以创建和更新帖子。
帖子可以包含标签。
每个标签都有自己的 feed，其中包含来自该标签的所有帖子。
所有的帖子都存储在 MySQL 中。这就是写模型。
所有 feed 都异步更新并存储在 MongoDB 中。这是读模型。
为什么要使用单独的写和读模型?
对于这个示例应用程序，使用多语言持久性(两个数据库引擎)当然有些过头了。我们这样做是为了展示这个技术，以及如何很容易地将它应用到 Watermill。
专用的读模型对于具有高读/写比率的应用程序是一种有用的模式。所有写操作都被原子地应用到写模型(在我们的例子中是 MySQL)。事件处理程序异步更新读模型(我们使用 Mongo)。
读取模型中的数据可以按原样使用。也可以独立于写模型进行扩展。
请记住，要使用此模式，应用程序中必须接受最终的一致性。而且，在大多数用例中，您可能不需要使用它。务实!
SSE Router
SSERouter 来自 watermill-http。当创建一个新的路由器时，你需要传递一个上游订阅者。来自该订阅服务器的消息将触发通过 HTTP 推送更新。
在本例中，我们使用 NATS 作为 Pub/Sub，但这可以是 Watermill 支持的任何 Pub/Sub。
sseRouter, err := watermillHTTP.NewSSERouter(
watermillHTTP.SSERouterConfig{
UpstreamSubscriber: router.Subscriber,
ErrorHandler: watermillHTTP.DefaultErrorHandler,
},
router.Logger,
)
Stream Adapters(流适配器)
要使用 SSERouter，你需要准备一个带有两个方法的 StreamAdapter。
GetResponse 类似于标准的 HTTP 处理程序。修改现有的处理程序来匹配这个签名应该非常容易。
Validate 是一个额外的方法，它告诉我们是否应该为特定的 Message 推送更新。
type StreamAdapter interface {
// GetResponse returns the response to be sent back to client.
// Any errors that occur should be handled and written to `w`, returning false as `ok`.
GetResponse(w http.ResponseWriter, r *http.Request) (response interface{}, ok bool)
// Validate validates if the incoming message should be handled by this handler.
// Typically this involves checking some kind of model ID.
Validate(r *http.Request, msg *message.Message) (ok bool)
}
Validate 示例如下所示。它检查消息是否来自与用户通过 HTTP 请求发送的相同的 post ID。
func (p postStreamAdapter) Validate(r *http.Request, msg *message.Message) (ok bool) {
postUpdated := PostUpdated{}
err := json.Unmarshal(msg.Payload, &postUpdated)
if err != nil {
return false
}
postID := chi.URLParam(r, "id")
return postUpdated.OriginalPost.ID == postID
}
如果你想为每条消息触发一个更新，你可以简单地返回 true。
func (f allFeedsStreamAdapter) Validate(r *http.Request, msg *message.Message) (ok bool) {
return true
}
在开始 SSERouter 之前，您需要添加带有特定主题的处理程序。 AddHandler 返回一个可以在任何路由库中使用的标准 HTTP 处理程序。
postHandler := sseRouter.AddHandler(PostUpdatedTopic, postStream)
// ...
r.Get("/posts/{id}", postHandler)
Event handlers(事件处理程序)
该示例使用 Watermill 进行所有异步通信，包括 SSE。
发布了以下事件:
PostCreated将 post 添加到贴子中包含标签的所有 feeds 中。
FeedUpdated将更新推送到当前访问 feed 页面的所有客户端。
PostUpdateda) 对于现有标签，帖子内容将在标签中更新。
b) 如果添加了新的标签，文章将被添加到标签的 feed 中。
c) 如果标签已删除，则该帖子将从标签的 feed 中删除。
将更新推送给所有当前访问 post 页面的客户端。
使用帖子中存在的标签更新所有 feeds 中的帖子
前端 app
前端应用程序是使用 Vue.js 和 Bootstrap 构建的。
最有趣的部分是 EventSource 的使用。
this.es = new EventSource('/api/feeds/' + this.feed)
this.es.addEventListener('data', event => {
let data = JSON.parse(event.data);
this.posts_stream = data.posts;
}, false);
Refs
watermill.io
