home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Redis

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Project Description - รายละเอียดโปรเจกต์

Blog Post นี้สร้าง Messaging Application ให้ Users สร้าง Message Clients และ Chat Room เข้าถึง Past Messages ได้

Project มี 2 Pages หน้าแรก Client Registration สร้าง Multiple Clients ชื่อ Unique

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Click Client Username → Directed ไป Chatroom Client ของ User นั้น

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Logic ของ Chat Application:

Users สร้าง Multiple Clients บน Index Page แต่ละตัว Unique Username Click Username → New Tab Client แยก Path Unique

แต่ละ Client เชื่อม Message Server ผ่าน WebSocket Connection Client สร้าง Message ใหม่ → ส่งไป Message Server ของ Client นั้น

Message Servers Handle Message Traffic Client ส่ง Message ผ่าน WebSocket → Server Direct ไป Kafka Broker แต่ละ Message Server รัน NodeJS Thread Handle Incoming Messages Consume Message → ส่งไป Clients ผ่าน WebSocket Connection ใช้ react-use-websocket Library Consume Incoming Messages Client-side

Application ใช้ Upstash Redis เก็บ Message History Produce Message ไป Kafka → Persist ไป Redis Database สร้าง Client ใหม่ → Retrieve Old Messages จาก Upstash Redis Render ใน Chat Display

Overview แอปพลิเคชัน:

Note: Implementation สร้าง Single Message Server สำหรับ Demo เพิ่ม Servers ได้เพื่อ Handle Load

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Demo - สาธิต

Demo App ที่นี่ Current Version Deploy บน Fly

Getting Started - เริ่มต้น

Steps สร้าง Chat Application:

  1. Creating Upstash Redis Database
  2. Creating Upstash Kafka Cluster
  3. Creating Next Application (Frontend)
  4. Creating WebSocket Message Server
  5. Deploying to Fly.io

Creating Upstash Redis Database

Upstash Console Login Tab Redis กด Create Database

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Redis Ready! กลับมาหา Credentials ใน Redis Console

Creating Upstash Kafka Cluster

Switch Tab Kafka กด Create Cluster ตั้งชื่อ Cluster Create Kafka Topic Confirm

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Creating the Next App

สร้าง Root Folder App จาก Terminal เก็บ Next App และ Server ใน Folder เดียวกัน:

mkdir chat-app
cd chat-app

สร้าง Next App:

$ npx create-next-app@latest
✔ What is your project named? … next-chat-app
✔ Would you like to use TypeScript? … Yes
✔ Would you like to use ESLint? … Yes
✔ Would you like to use Tailwind CSS? … No
✔ Would you like to use `src/` directory? … No
✔ Would you like to use App Router? (recommended) … No
✔ Would you like to customize the default import alias? … No

Handling Credentials - จัดการ Credentials

สร้าง .env เก็บ Credentials Import จากไฟล์นี้ ไม่ต้อง Copy Paste ซ้ำ

สร้าง .env ไป Redis Console Copy/Paste UPSTASH_REDIS_REST_URL และ UPSTASH_REDIS_REST_TOKEN ใส่ .env

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

Switch Kafka Console Transfer UPSTASH_KAFKA_REST_URL, UPSTASH_KAFKA_REST_USERNAME, UPSTASH_KAFKA_REST_PASSWORD

สร้าง Real-Time Chat App ด้วย Upstash Kafka, Redis และ Next.js

.env ควรเป็นแบบนี้:

UPSTASH_REDIS_REST_URL=...
UPSTASH_REDIS_REST_TOKEN=...
UPSTASH_KAFKA_REST_URL=...
UPSTASH_KAFKA_REST_USERNAME=...
UPSTASH_KAFKA_REST_PASSWORD=...

Client Registration Page - หน้าลงทะเบียน Client

Index Page Client Registration/Creation Submit Username → New Client Created Listed Under Current Clients Table

import { useState } from "react";
import Link from "next/link";
import { Redis } from "@upstash/redis";
import styles from "@/styles/Home.module.css";
export default function Home() {
  const [usernameInput, setUsernameInput] = useState<string>("");
  const [usernameList, setUsernameList] = useState<string[]>(Array<string>);
  const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>): void => { const inputValue: string = e.target.value; setUsernameInput(inputValue); };
  const addUsernameClient = (e: React.FormEvent<HTMLFormElement>): void => { e.preventDefault(); setUsernameList([...usernameList, usernameInput]); setUsernameInput(""); };
  return (
    <div className={styles.container}>
      <div className={styles.welcomeSection}>
        <h1>Welcome to the demo message app!</h1>
        <p> This application uses Upstash Kafka for message passing, and Upstash Redis for state management. <br /> <br /> To get started, create several clients by typing in unique usernames to the input section below and submitting. <br /> <br /> The usernames will be added to the list of current clients. Click on a username to open a new tab with that client&apos;s message display. <br /> <br /> You can have multiple sessions open at once. </p>
      </div>
      <form className={styles.formSection} onSubmit={addUsernameClient}>
        <input type="text" className={styles.formInput} value={usernameInput} onChange={handleInputChange} ></input>
        <button className={styles.formSubmit} type="submit"> Create the client! </button>
      </form>
      <div className={styles.clientListSection}>
        <p className={styles.clientListHeader}>Current Clients</p>
        <div className={styles.clientList}>
          {usernameList.map((username, i) => { return ( <Link href={`/user/${username}`} key={`${i}`} className={styles.userClient} target={"_blank"} > <p>{username}</p> </Link> ); })} </div>
      </div>
    </div>
  );
}

Reset Chat History ทุกครั้ง Reload App ใช้ Function นี้:

export async function getServerSideProps() {
  const redis = new Redis({ url: process.env.UPSTASH_REDIS_REST_URL, token: process.env.UPSTASH_REDIS_REST_TOKEN });
  await redis.del("messagesList");
  return { props: {} };
}

Index Page Ready รัน npm run dev ใน next-chat-app Folder ดู Index Page Live

Message Client Page - หน้า Message Client

Dynamic Routing สร้าง Folder /pages/user/[username].tsx Structure นี้ Dynamic Routes ต่อ Client ตาม Username

Main Component Client Hold States Message List, Username ใช้ useWebSocket Hook สร้าง Message, Connection, Disconnection Events จาก WebSocket Message Event Emit → Add Message List → MessageDisplay Rerender

import { useState } from "react";
import { useRouter } from "next/router";
import { Redis } from "@upstash/redis";
import useWebSocket from "react-use-websocket";
import styles from "@/styles/Home.module.css";
type Message = { id: number; sender: string; text: string; };
export default function MessageApp(props: { messagesData: Message[] }) {
  const { messagesData } = props;
  const { username } = useRouter().query;
  const [inputText, setInputText] = useState<string>("");
  const [messageList, setMessageList] = useState<Message[]>(messagesData);
  const [messageCounter, setMessageCounter] = useState<number>(0);
  const handleMessage = function (message: Message) { const nextMessages = [...messageList, message]; setMessageList(nextMessages); };
  const { sendMessage } = useWebSocket("ws://localhost:8080", { share: true, filter: () => false, onOpen: () => { console.log("WebSocket connection!"); return "connection"; }, onMessage: (message) => { const data = JSON.parse(message.data); const { sender, text }: { sender: string; text: string } = data; const messageData: Message = { id: messageCounter, sender: sender, text: text }; setMessageCounter(messageCounter + 1); handleMessage(messageData); return message; }, onClose: () => { console.log("WebSocket disconnected!"); return "disconnected"; } });
  function handleSendMessage(messageText: string) { const messageData = { sender: username, text: messageText }; sendMessage(JSON.stringify(messageData)); }
  return ( <div className={styles.Container}> <MessageDisplay messages={messageList} /> <MessageInput inputText={inputText} setInputText={setInputText} handleSendMessage={handleSendMessage} /> </div> );
}

MessageDisplay และ MessageInput Components:

const MessageDisplay = function (props: { messages: Message[] }) { const { messages } = props; return ( <div className={styles.messageContainer}> {messages.map((message) => ( <MessageBubble key={message.id} sender={message.sender} text={message.text} /> ))} </div> ); };
const MessageInput = (props: { inputText: string; setInputText: (msg: string) => void; handleSendMessage: (msg: string) => void; }) => { const { inputText, setInputText, handleSendMessage } = props; const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>): void => { const inputValue: string = e.target.value; setInputText(inputValue); }; const handleSubmit = (e: React.FormEvent<HTMLFormElement>): void => { e.preventDefault(); handleSendMessage(inputText); if (inputText.trim() !== "") { setInputText(" "); } }; return ( <form className={styles.inputSection} onSubmit={handleSubmit}> <input className={styles.inputText} type="text" value={inputText} onChange={handleInputChange} ></input> <button className={styles.inputSendButton} type="submit"> Send </button> </form> ); };
const MessageBubble = (props: { sender: string; text: string; key: number; }) => { const { sender, text } = props; const { username } = useRouter().query; const isSender = sender === username; const senderClass = isSender ? "sender" : "receiver"; return ( <div className={`${styles["messageBubble"]} ${styles[senderClass]}`}> <div className={styles.messageSender}> {isSender ? "You" : sender} </div> <div className={styles.messageText}>{text}</div> </div> ); };

Provide Chat History ให้ Client ใช้ getServerSideProps():

export async function getServerSideProps() {
  const redis = new Redis({ url: process.env.UPSTASH_REDIS_REST_URL, token: process.env.UPSTASH_REDIS_REST_TOKEN });
  const messagesData = (await redis.lrange("messagesList", 0, -1)).reverse();
  return { props: { messagesData } };
}

Next.js App Working Refresh Page Create Clients Navigate หนึ่งในนั้น จะเห็น Client Page แต่ยังต้อง Message Servers Handle Message Flow

Creating Message Server - สร้าง Message Server

Server Structure เรียบง่าย ใช้ Node.js, ws Library, Upstash Kafka สร้าง Folder server ใน chat-app folder:

mkdir server
cd server

ใน server Folder Install Requirements Config Files:

npm install typescript ws tsc @upstash/kafka @types/ws tsc --init

สร้าง WebSocket, Kafka Producer, Kafka Consumer Clients ใน /server/message_server.ts:

import * as http from "http";
import { Kafka } from "@upstash/kafka";
import { Redis } from "@upstash/redis";
import { WebSocket } from "ws";
const server = http.createServer();
const wss = new WebSocket.Server({ server });
server.listen(8080, () => { console.log("Server is running on port 8080"); });
const kafka = new Kafka({ url: process.env.UPSTASH_KAFKA_REST_URL, username: process.env.UPSTASH_KAFKA_REST_USERNAME, password: process.env.UPSTASH_KAFKA_REST_PASSWORD });
const redis = new Redis({ url: process.env.UPSTASH_REDIS_REST_URL, token: process.env.UPSTASH_REDIS_REST_TOKEN });
const consumer = kafka.consumer();
const producer = kafka.producer();
const clients = new Set<WebSocket>();

Interact WebSocket สร้าง connection และ message Events:

wss.on("connection", async (connection, req) => {
  clients.add(connection);
  console.log(`New client connected!`);
  connection.on("message", async (message) => {
    const jsonMessage = message.toString();
    console.log("Received message:", JSON.parse(jsonMessage));
    producer.produce("chat", jsonMessage);
  });
  connection.on("close", () => {
    console.log(`Client disconnected:`);
    clients.delete(connection);
  });
});

สร้างและรัน Thread Consume Messages ด้วย Interval:

async function run() {
  while (true) {
    const messages = await consumer.consume({ consumerGroupId: "group_1", instanceId: "instance_1", topics: ["chat"], autoOffsetReset: "earliest" });
    if (messages.length != 0) {
      for (let i = 0; i < messages.length; i++) {
        await redis.lpush("messagesList", messages[i].value);
        console.log(`Message sending: ${messages[i].value}`);
        clients.forEach((connection: WebSocket) => { connection.send(messages[i].value); });
      }
    }
    console.log("Run!");
    await new Promise((resolve) => setTimeout(resolve, 1000));
  }
}

Everything Ready App ควรทำงานได้ Run Message Server Local Refresh Client Page ดู Messages Transmitted ระหว่าง Clients Commands Transcompile TS และ Run Server localhost:8000:

tsc message_server.ts
node message_server.js

Deployment - การ Deploy

ใช้ Fly.io สร้าง Account ก่อนเริ่ม

Deploying the Message Server - Deploy Message Server

ไป server Folder Install flyctl CLI Tool Authorize via Shell:

npm install flyctl
flyctl auth login

สร้าง Config Files flyctl init สร้าง fly.toml Edit fly.toml ใส่ WebSocket Connection Configs:

[[services]]
internal_port = 8080
protocol = "tcp"
[services.concurrency]
hard_limit = 25
soft_limit = 20
[[services.ports]]
handlers = ["http"]
port = "80"
[[services.ports]]
handlers = ["tls", "http"]
port = "443"
[[services.tcp_checks]]
interval = 10000
timeout = 2000

Final Step Servers รัน flyctl deploy พร้อม! Deploy เสร็จ flyctl ให้ Endpoint Server Copy Endpoint ในกรณีนี้ message-server.fly.dev

Deploying the Next Application - Deploy Next Application

ก่อน Deploy Next.js Embed Deployment Endpoint Message Server Replace WebSocket URL ใน pages/user/[username].tsx จาก ws://localhost:8080 เป็น Endpoint จาก flyctl + wss:// Prefix กรณีนี้ wss://message-server.fly.dev

ใน next-chat-app Folder รัน Commands เหมือน server ไม่ต้อง Edit fly.toml ข้าม Step นั้น:

flyctl init
flyctl deploy

เสร็จแล้ว! flyctl open Navigate ไป Deployed Project

Conclusion and Suggestions - สรุปและข้อเสนอแนะ

ขอบคุณที่ติดตาม!

GitHub Repository ที่นี่

อยากต่อยอดโปรเจกต์ ข้อเสนอ:

  • ปัจจุบัน Reload Page → Flush Stored Messages ใน Upstash Redis Controlled by Code ใน pages/index.tsx getServerSideProps Function Issue: User Reload → Delete Chat History ทุกคนใน Chatroom แก้: Implement TTL Extension สำหรับ Chat History ทุกครั้งส่ง Message จะ Preserve History หลัง Reload
  • Implement Multiple Chatrooms Feature สร้าง Multiple Kafka Topics Unique Names ต่อ Chatroom หรือ Handle บน Message Server ใช้ Data Structures ที่เหมาะสม
  • Implement Multiple Message Servers และ Load Balancer Apply Best System Design Practices

คำถาม ติดต่อ fahreddin@upstash.com