构建一个即时消息应用（四）：消息

本文是该系列的第四篇。

  • 第一篇: 模式
  • 第二篇: OAuth
  • 第三篇: 对话

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

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

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

创建消息

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

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

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

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

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

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

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

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

如果用户不是对话参与者，我们将返回一个 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.   
  
  
  2.  if ctx == nil {
    3.   
  
  
  3.  ctx = context.Background()
    4.   
  
  
  4.  }
    5.   
  
  
  5.  
    6.   
  
  
  6.  if _, err := db.ExecContext(ctx, `
    7.   
  
  
  7.  UPDATE participants SET messages_read_at = now()
    8.   
  
  
  8.  WHERE user_id = $1 AND conversation_id = $2
    9.   
  
  
  9.  `, userID, conversationID); err != nil {
    10.   
  
  
  10.  return err
    11.   
  
  
  11.  }
    12.   
  
  
  12.  return nil
    13.   
  
  
  13. }
    14.

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

获取消息

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

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

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

读取消息

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

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

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

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

  • 源代码

分享题目：构建一个即时消息应用（四）：消息
网页路径：http://www.shufengxianlan.com/qtweb/news29/370979.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

品牌网站制作知识

各行业网站

租用服务器　　 　大英泰恒网站　　 　电信成都枢纽中心　　 　重庆服务器托管　　 　响应式网站　　 　四川成都网站建设　　 　seo优化排名　　 　成都网络推广　　 　成都花箱　　 　子晨包装　　 　虚拟主机　　 　bgp机房托管　　 　成都小谭网创广告　　 　微信营销　　 　云服务器　　 　关键词优化排名　　 　成都网站制作　　 　成都多线服务器托管　　 　移动机房服务器托管　　 　专业网站设计