Linux中國

構建一個即時消息應用（四）：消息

131 views0
By
Share

本文是該系列的第四篇。

在這篇文章中，我們將對端點進行編碼，以創建一條消息並列出它們，同時還將編寫一個端點以更新參與者上次閱讀消息的時間。 首先在 main() 函數中添加這些路由。

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

消息會進入對話，因此端點包含對話 ID。

創建消息

該端點處理對 /api/conversations/{conversationID}/messages 的 POST 請求，其 JSON 主體僅包含消息內容，並返回新創建的消息。它有兩個副作用：更新對話 last_message_id 以及更新參與者 messages_read_at。

func createMessage(w http.ResponseWriter, r *http.Request) {
    var input struct {
        Content string `json:"content"`
    }
    defer r.Body.Close()
    if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    errs := make(map[string]string)
    input.Content = removeSpaces(input.Content)
    if input.Content == "" {
        errs["content"] = "Message content required"
    } else if len([]rune(input.Content)) > 480 {
        errs["content"] = "Message too long. 480 max"
    }
    if len(errs) != 0 {
        respond(w, Errors{errs}, http.StatusUnprocessableEntity)
        return
    }

    ctx := r.Context()
    authUserID := ctx.Value(keyAuthUserID).(string)
    conversationID := way.Param(ctx, "conversationID")

    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        respondError(w, fmt.Errorf("could not begin tx: %v", err))
        return
    }
    defer tx.Rollback()

    isParticipant, err := queryParticipantExistance(ctx, tx, authUserID, conversationID)
    if err != nil {
        respondError(w, fmt.Errorf("could not query participant existance: %v", err))
        return
    }

    if !isParticipant {
        http.Error(w, "Conversation not found", http.StatusNotFound)
        return
    }

    var message Message
    if err := tx.QueryRowContext(ctx, `
        INSERT INTO messages (content, user_id, conversation_id) VALUES
            ($1, $2, $3)
        RETURNING id, created_at
    `, input.Content, authUserID, conversationID).Scan(
        &message.ID,
        &message.CreatedAt,
    ); err != nil {
        respondError(w, fmt.Errorf("could not insert message: %v", err))
        return
    }

    if _, err := tx.ExecContext(ctx, `
        UPDATE conversations SET last_message_id = $1
        WHERE id = $2
    `, message.ID, conversationID); err != nil {
        respondError(w, fmt.Errorf("could not update conversation last message ID: %v", err))
        return
    }

    if err = tx.Commit(); err != nil {
        respondError(w, fmt.Errorf("could not commit tx to create a message: %v", err))
        return
    }

    go func() {
        if err = updateMessagesReadAt(nil, authUserID, conversationID); err != nil {
            log.Printf("could not update messages read at: %vn", err)
        }
    }()

    message.Content = input.Content
    message.UserID = authUserID
    message.ConversationID = conversationID
    // TODO: notify about new message.
    message.Mine = true

    respond(w, message, http.StatusCreated)
}

首先，它將請求正文解碼為包含消息內容的結構。然後，它驗證內容不為空並且少於 480 個字元。

var rxSpaces = regexp.MustCompile("\s+")

func removeSpaces(s string) string {
    if s == "" {
        return s
    }

    lines := make([]string, 0)
    for _, line := range strings.Split(s, "n") {
        line = rxSpaces.ReplaceAllLiteralString(line, " ")
        line = strings.TrimSpace(line)
        if line != "" {
            lines = append(lines, line)
        }
    }
    return strings.Join(lines, "n")
}

這是刪除空格的函數。它遍歷每一行，刪除兩個以上的連續空格，然後回非空行。

驗證之後，它將啟動一個 SQL 事務。首先，它查詢對話中的參與者是否存在。

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

我將其提取到一個函數中，因為稍後可以重用。

如果用戶不是對話參與者，我們將返回一個 404 NOT Found 錯誤。

然後，它插入消息並更新對話 last_message_id。從這時起，由於我們不允許刪除消息，因此 last_message_id 不能為 NULL。

接下來提交事務，並在 goroutine 中更新參與者 messages_read_at。

func updateMessagesReadAt(ctx context.Context, userID, conversationID string) error {
    if ctx == nil {
        ctx = context.Background()
    }

    if _, err := db.ExecContext(ctx, `
        UPDATE participants SET messages_read_at = now()
        WHERE user_id = $1 AND conversation_id = $2
    `, userID, conversationID); err != nil {
        return err
    }
    return nil
}

在回復這條新消息之前，我們必須通知一下。這是我們將要在下一篇文章中編寫的實時部分，因此我在那裡留一了個注釋。

獲取消息

這個端點處理對 /api/conversations/{conversationID}/messages 的 GET 請求。 它用一個包含會話中所有消息的 JSON 數組進行響應。它還具有更新參與者 messages_read_at 的副作用。

func getMessages(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    authUserID := ctx.Value(keyAuthUserID).(string)
    conversationID := way.Param(ctx, "conversationID")

    tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
    if err != nil {
        respondError(w, fmt.Errorf("could not begin tx: %v", err))
        return
    }
    defer tx.Rollback()

    isParticipant, err := queryParticipantExistance(ctx, tx, authUserID, conversationID)
    if err != nil {
        respondError(w, fmt.Errorf("could not query participant existance: %v", err))
        return
    }

    if !isParticipant {
        http.Error(w, "Conversation not found", http.StatusNotFound)
        return
    }

    rows, err := tx.QueryContext(ctx, `
        SELECT
            id,
            content,
            created_at,
            user_id = $1 AS mine
        FROM messages
        WHERE messages.conversation_id = $2
        ORDER BY messages.created_at DESC
    `, authUserID, conversationID)
    if err != nil {
        respondError(w, fmt.Errorf("could not query messages: %v", err))
        return
    }
    defer rows.Close()

    messages := make([]Message, 0)
    for rows.Next() {
        var message Message
        if err = rows.Scan(
            &message.ID,
            &message.Content,
            &message.CreatedAt,
            &message.Mine,
        ); err != nil {
            respondError(w, fmt.Errorf("could not scan message: %v", err))
            return
        }

        messages = append(messages, message)
    }

    if err = rows.Err(); err != nil {
        respondError(w, fmt.Errorf("could not iterate over messages: %v", err))
        return
    }

    if err = tx.Commit(); err != nil {
        respondError(w, fmt.Errorf("could not commit tx to get messages: %v", err))
        return
    }

    go func() {
        if err = updateMessagesReadAt(nil, authUserID, conversationID); err != nil {
            log.Printf("could not update messages read at: %vn", err)
        }
    }()

    respond(w, messages, http.StatusOK)
}

首先，它以只讀模式開始一個 SQL 事務。檢查參與者是否存在，並查詢所有消息。在每條消息中，我們使用當前經過身份驗證的用戶 ID 來了解用戶是否擁有該消息（mine）。 然後，它提交事務，在 goroutine 中更新參與者 messages_read_at 並以消息響應。

讀取消息

該端點處理對 /api/conversations/{conversationID}/read_messages 的 POST 請求。 沒有任何請求或響應主體。 在前端，每次有新消息到達實時流時，我們都會發出此請求。

func readMessages(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    authUserID := ctx.Value(keyAuthUserID).(string)
    conversationID := way.Param(ctx, "conversationID")

    if err := updateMessagesReadAt(ctx, authUserID, conversationID); err != nil {
        respondError(w, fmt.Errorf("could not update messages read at: %v", err))
        return
    }

    w.WriteHeader(http.StatusNoContent)
}

它使用了與更新參與者 messages_read_at 相同的函數。

到此為止。實時消息是後台僅剩的部分了。請等待下一篇文章。

via: https://nicolasparada.netlify.com/posts/go-messenger-messages/

作者：Nicolás Parada 選題：lujun9972 譯者：gxlct008 校對：wxy

本文由 LCTT 原創編譯，Linux中國 榮譽推出

本文轉載來自 Linux 中國: https://github.com/Linux-CN/archive

對這篇文章感覺如何？

太棒了
0
不錯
0
愛死了
0
不太好
0
感覺很糟
0
雨落清風。心向陽
    Previous
    Next

    You may also like

    Leave a reply

    您的電子郵箱地址不會被公開。 必填項已用 * 標註

    此站點使用Akismet來減少垃圾評論。了解我們如何處理您的評論數據。

    More in:Linux中國