文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

构建一个即时消息应用(四):消息

2024-12-10 15:28

关注

本文是该系列的第四篇。

在这篇文章中,我们将对端点进行编码,以创建一条消息并列出它们,同时还将编写一个端点以更新参与者上次阅读消息的时间。 首先在 main() 函数中添加这些路由。

  1. router.HandleFunc("POST", "/api/conversations/:conversationID/messages", requireJSON(guard(createMessage)))
  2. router.HandleFunc("GET", "/api/conversations/:conversationID/messages", guard(getMessages))
  3. router.HandleFunc("POST", "/api/conversations/:conversationID/read_messages", guard(readMessages))

消息会进入对话,因此端点包含对话 ID。

创建消息

该端点处理对 /api/conversations/{conversationID}/messages 的 POST 请求,其 JSON 主体仅包含消息内容,并返回新创建的消息。它有两个副作用:更新对话 last_message_id 以及更新参与者 messages_read_at

  1. func createMessage(w http.ResponseWriter, r *http.Request) {
  2. var input struct {
  3. Content string `json:"content"`
  4. }
  5. defer r.Body.Close()
  6. if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
  7. http.Error(w, err.Error(), http.StatusBadRequest)
  8. return
  9. }
  10.  
  11. errs := make(map[string]string)
  12. input.Content = removeSpaces(input.Content)
  13. if input.Content == "" {
  14. errs["content"] = "Message content required"
  15. } else if len([]rune(input.Content)) > 480 {
  16. errs["content"] = "Message too long. 480 max"
  17. }
  18. if len(errs) != 0 {
  19. respond(w, Errors{errs}, http.StatusUnprocessableEntity)
  20. return
  21. }
  22.  
  23. ctx := r.Context()
  24. authUserID := ctx.Value(keyAuthUserID).(string)
  25. conversationID := way.Param(ctx, "conversationID")
  26.  
  27. tx, err := db.BeginTx(ctx, nil)
  28. if err != nil {
  29. respondError(w, fmt.Errorf("could not begin tx: %v", err))
  30. return
  31. }
  32. defer tx.Rollback()
  33.  
  34. isParticipant, err := queryParticipantExistance(ctx, tx, authUserID, conversationID)
  35. if err != nil {
  36. respondError(w, fmt.Errorf("could not query participant existance: %v", err))
  37. return
  38. }
  39.  
  40. if !isParticipant {
  41. http.Error(w, "Conversation not found", http.StatusNotFound)
  42. return
  43. }
  44.  
  45. var message Message
  46. if err := tx.QueryRowContext(ctx, `
  47. INSERT INTO messages (content, user_id, conversation_id) VALUES
  48. ($1, $2, $3)
  49. RETURNING id, created_at
  50. `, input.Content, authUserID, conversationID).Scan(
  51. &message.ID,
  52. &message.CreatedAt,
  53. ); err != nil {
  54. respondError(w, fmt.Errorf("could not insert message: %v", err))
  55. return
  56. }
  57.  
  58. if _, err := tx.ExecContext(ctx, `
  59. UPDATE conversations SET last_message_id = $1
  60. WHERE id = $2
  61. `, message.ID, conversationID); err != nil {
  62. respondError(w, fmt.Errorf("could not update conversation last message ID: %v", err))
  63. return
  64. }
  65.  
  66. if err = tx.Commit(); err != nil {
  67. respondError(w, fmt.Errorf("could not commit tx to create a message: %v", err))
  68. return
  69. }
  70.  
  71. go func() {
  72. if err = updateMessagesReadAt(nil, authUserID, conversationID); err != nil {
  73. log.Printf("could not update messages read at: %v\n", err)
  74. }
  75. }()
  76.  
  77. message.Content = input.Content
  78. message.UserID = authUserID
  79. message.ConversationID = conversationID
  80. // TODO: notify about new message.
  81. message.Mine = true
  82.  
  83. respond(w, message, http.StatusCreated)
  84. }

首先,它将请求正文解码为包含消息内容的结构。然后,它验证内容不为空并且少于 480 个字符。

  1. var rxSpaces = regexp.MustCompile("\\s+")
  2.  
  3. func removeSpaces(s string) string {
  4. if s == "" {
  5. return s
  6. }
  7.  
  8. lines := make([]string, 0)
  9. for _, line := range strings.Split(s, "\n") {
  10. line = rxSpaces.ReplaceAllLiteralString(line, " ")
  11. line = strings.TrimSpace(line)
  12. if line != "" {
  13. lines = append(lines, line)
  14. }
  15. }
  16. return strings.Join(lines, "\n")
  17. }

这是删除空格的函数。它遍历每一行,删除两个以上的连续空格,然后回非空行。

验证之后,它将启动一个 SQL 事务。首先,它查询对话中的参与者是否存在。

  1. func queryParticipantExistance(ctx context.Context, tx *sql.Tx, userID, conversationID string) (bool, error) {
  2. if ctx == nil {
  3. ctx = context.Background()
  4. }
  5. var exists bool
  6. if err := tx.QueryRowContext(ctx, `SELECT EXISTS (
  7. SELECT 1 FROM participants
  8. WHERE user_id = $1 AND conversation_id = $2
  9. )`, userID, conversationID).Scan(&exists); err != nil {
  10. return false, err
  11. }
  12. return exists, nil
  13. }

我将其提取到一个函数中,因为稍后可以重用。

如果用户不是对话参与者,我们将返回一个 404 NOT Found 错误。

然后,它插入消息并更新对话 last_message_id。从这时起,由于我们不允许删除消息,因此 last_message_id 不能为 NULL

接下来提交事务,并在 goroutine 中更新参与者 messages_read_at

  1. func updateMessagesReadAt(ctx context.Context, userID, conversationID string) error {
  2. if ctx == nil {
  3. ctx = context.Background()
  4. }
  5.  
  6. if _, err := db.ExecContext(ctx, `
  7. UPDATE participants SET messages_read_at = now()
  8. WHERE user_id = $1 AND conversation_id = $2
  9. `, userID, conversationID); err != nil {
  10. return err
  11. }
  12. return nil
  13. }

在回复这条新消息之前,我们必须通知一下。这是我们将要在下一篇文章中编写的实时部分,因此我在那里留一了个注释。

获取消息

这个端点处理对 /api/conversations/{conversationID}/messages 的 GET 请求。 它用一个包含会话中所有消息的 JSON 数组进行响应。它还具有更新参与者 messages_read_at 的副作用。

  1. func getMessages(w http.ResponseWriter, r *http.Request) {
  2. ctx := r.Context()
  3. authUserID := ctx.Value(keyAuthUserID).(string)
  4. conversationID := way.Param(ctx, "conversationID")
  5.  
  6. tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
  7. if err != nil {
  8. respondError(w, fmt.Errorf("could not begin tx: %v", err))
  9. return
  10. }
  11. defer tx.Rollback()
  12.  
  13. isParticipant, err := queryParticipantExistance(ctx, tx, authUserID, conversationID)
  14. if err != nil {
  15. respondError(w, fmt.Errorf("could not query participant existance: %v", err))
  16. return
  17. }
  18.  
  19. if !isParticipant {
  20. http.Error(w, "Conversation not found", http.StatusNotFound)
  21. return
  22. }
  23.  
  24. rows, err := tx.QueryContext(ctx, `
  25. SELECT
  26. id,
  27. content,
  28. created_at,
  29. user_id = $1 AS mine
  30. FROM messages
  31. WHERE messages.conversation_id = $2
  32. ORDER BY messages.created_at DESC
  33. `, authUserID, conversationID)
  34. if err != nil {
  35. respondError(w, fmt.Errorf("could not query messages: %v", err))
  36. return
  37. }
  38. defer rows.Close()
  39.  
  40. messages := make([]Message, 0)
  41. for rows.Next() {
  42. var message Message
  43. if err = rows.Scan(
  44. &message.ID,
  45. &message.Content,
  46. &message.CreatedAt,
  47. &message.Mine,
  48. ); err != nil {
  49. respondError(w, fmt.Errorf("could not scan message: %v", err))
  50. return
  51. }
  52.  
  53. messages = append(messages, message)
  54. }
  55.  
  56. if err = rows.Err(); err != nil {
  57. respondError(w, fmt.Errorf("could not iterate over messages: %v", err))
  58. return
  59. }
  60.  
  61. if err = tx.Commit(); err != nil {
  62. respondError(w, fmt.Errorf("could not commit tx to get messages: %v", err))
  63. return
  64. }
  65.  
  66. go func() {
  67. if err = updateMessagesReadAt(nil, authUserID, conversationID); err != nil {
  68. log.Printf("could not update messages read at: %v\n", err)
  69. }
  70. }()
  71.  
  72. respond(w, messages, http.StatusOK)
  73. }

首先,它以只读模式开始一个 SQL 事务。检查参与者是否存在,并查询所有消息。在每条消息中,我们使用当前经过身份验证的用户 ID 来了解用户是否拥有该消息(mine)。 然后,它提交事务,在 goroutine 中更新参与者 messages_read_at 并以消息响应。

读取消息

该端点处理对 /api/conversations/{conversationID}/read_messages 的 POST 请求。 没有任何请求或响应主体。 在前端,每次有新消息到达实时流时,我们都会发出此请求。

  1. func readMessages(w http.ResponseWriter, r *http.Request) {
  2. ctx := r.Context()
  3. authUserID := ctx.Value(keyAuthUserID).(string)
  4. conversationID := way.Param(ctx, "conversationID")
  5.  
  6. if err := updateMessagesReadAt(ctx, authUserID, conversationID); err != nil {
  7. respondError(w, fmt.Errorf("could not update messages read at: %v", err))
  8. return
  9. }
  10.  
  11. w.WriteHeader(http.StatusNoContent)
  12. }

它使用了与更新参与者 messages_read_at 相同的函数。


到此为止。实时消息是后台仅剩的部分了。请等待下一篇文章。

 

来源:51CTO内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯