Upstash Kafka, Redis, और Next.js के साथ एक रियल-टाइम चैट ऐप बनाएं

प्रोजेक्ट विवरण

इस ब्लॉग पोस्ट में, हम एक मैसेजिंग एप्लिकेशन बना रहे हैं जो उपयोगकर्ताओं को मैसेज क्लाइंट और चैट रूम बनाने की अनुमति देता है। इसके अतिरिक्त, उपयोगकर्ता पिछले संदेशों तक पहुंच सकेंगे।

प्रोजेक्ट में दो पेज होते हैं। पहला पेज क्लाइंट पंजीकरण के लिए समर्पित है, जहां आप अद्वितीय नामों के साथ कई क्लाइंट बना सकते हैं।

जब आप किसी क्लाइंट के उपयोगकर्ता नाम पर क्लिक करते हैं, तो आपको उस विशिष्ट उपयोगकर्ता से जुड़े चैटरूम क्लाइंट पर निर्देशित किया जाएगा।

चैट एप्लिकेशन का तर्क इस प्रकार है:

उपयोगकर्ता इंडेक्स पेज पर कई क्लाइंट बना सकते हैं, प्रत्येक एक अद्वितीय उपयोगकर्ता नाम के साथ। किसी क्लाइंट के उपयोगकर्ता नाम पर क्लिक करने से उस क्लाइंट के साथ एक अद्वितीय पथ के साथ एक नए टैब में एक अलग क्लाइंट खुल जाएगा।

प्रत्येक क्लाइंट एक WebSocket कनेक्शन के माध्यम से एक मैसेज सर्वर से जुड़ा होगा। जब क्लाइंट पर एक नया संदेश बनाया जाता है, तो इसे उस क्लाइंट से जुड़े मैसेज सर्वर पर भेजा जाएगा।

मैसेज सर्वर संदेश ट्रैफिक को संभालेंगे। जब कोई क्लाइंट WebSocket कनेक्शन के माध्यम से एक संदेश भेजता है, तो सर्वर इसे एक Kafka ब्रोकर पर निर्देशित करेगा। प्रत्येक मैसेज सर्वर आने वाले संदेशों को संभालने के लिए एक NodeJS थ्रेड चलाएगा। जब एक संदेश उपभोग किया जाता है, तो इसे मौजूदा WebSocket कनेक्शन के माध्यम से क्लाइंट्स को भेजा जाएगा। क्लाइंट-साइड पर आने वाले संदेशों का उपभोग करने के लिए, हम react-use-websocket लाइब्रेरी का उपयोग करेंगे।

एप्लिकेशन संदेश इतिहास संग्रहीत करने के लिए Upstash Redis का उपयोग करेगा। जब एक संदेश Kafka पर उत्पन्न होता है, तो इसे Redis डेटाबेस में भी बनाए रखा जाएगा। एक नया क्लाइंट बनाने पर, पुराने संदेशों को Upstash Redis से पुनः प्राप्त किया जाएगा और चैट डिस्प्ले में रेंडर किया जाएगा।

यहां एप्लिकेशन का सामान्य अवलोकन है:

नोट: हमारे कार्यान्वयन में, हम डेमो उद्देश्यों के लिए एक एकल मैसेज सर्वर बनाएंगे, कोई संदेश लोड को संभालने के लिए सर्वर की संख्या बढ़ा सकता है।

डेमो

आप ऐप का डेमो यहां देख सकते हैं। एप्लिकेशन का वर्तमान संस्करण Fly पर डिप्लॉयड है।

शुरुआत करना

चैट एप्लिकेशन बनाने के चरण यहां दिए गए हैं:

1. Upstash Redis डेटाबेस बनाना
2. Upstash Kafka क्लस्टर बनाना
3. Next एप्लिकेशन बनाना (फ्रंटएंड)।
4. WebSocket मैसेज सर्वर बनाना।
5. Fly.io पर एप्लिकेशन डिप्लॉय करना

Upstash Redis डेटाबेस बनाना

Upstash कंसोल पर नेविगेट करें और लॉग इन करें, फिर Redis टैब पर, Create Database बटन पर क्लिक करें।

बस, हमारा Redis उपयोग के लिए तैयार है! हम क्रेडेंशियल्स के लिए Redis कंसोल पर वापस आएंगे।

Upstash Kafka क्लस्टर बनाना

अब, Kafka टैब पर स्विच करें, और Create Cluster बटन पर क्लिक करें। क्लस्टर नाम टाइप करें और आगे बढ़ें। फिर, Kafka टॉपिक बनाएं और पुष्टि करें।

Next ऐप बनाना

सबसे पहले, अपने टर्मिनल से अपने एप्लिकेशन के रूट फ़ोल्डर को बनाएं और नेविगेट करें। हम इस फ़ोल्डर में Next ऐप और सर्वर दोनों को रखेंगे।

mkdir chat-app cd chat-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

क्रेडेंशियल्स संभालना

हम क्रेडेंशियल स्टोर करने के लिए .env नामक एक फ़ाइल बनाएंगे। हमें क्रेडेंशियल को बार-बार कॉपी और पेस्ट नहीं करना पड़ेगा, हम बस इस फ़ाइल से आयात करेंगे।

पहले, .env फ़ाइल बनाएं।

फिर, Redis कंसोल पर नेविगेट करें, और UPSTASH_REDIS_REST_URL और UPSTASH_REDIS_REST_TOKEN क्रेडेंशियल को .env फ़ाइल में कॉपी/पेस्ट करें।

अंत में, Kafka कंसोल पर स्विच करें, और UPSTASH_KAFKA_REST_URL, UPSTASH_KAFKA_REST_USERNAME, UPSTASH_KAFKA_REST_PASSWORD स्थानांतरित करें

अब, आपकी .env फ़ाइल इसी तरह दिखनी चाहिए

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

अब जब हमने क्रेडेंशियल कॉन्फ़िगर कर लिए हैं, तो हम एप्लिकेशन के साथ आगे बढ़ सकते हैं।

क्लाइंट पंजीकरण पेज

इंडेक्स पेज में क्लाइंट पंजीकरण/निर्माण ऑपरेशन होंगे। जब एक उपयोगकर्ता नाम सबमिट किया जाता है, तो एक नया क्लाइंट बनाया जाएगा और Current Clients तालिका के तहत सूचीबद्ध किया जाएगा।

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> ); }

यदि आप हर बार ऐप रीलोड होने पर चैट इतिहास रीसेट करना चाहते हैं, तो आप निम्न फ़ंक्शन का उपयोग कर सकते हैं:

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: {}, }; }

इसके साथ, इंडेक्स पेज चलाने के लिए तैयार है। next-chat-app फ़ोल्डर में npm run dev कमांड चलाएं ताकि इंडेक्स पेज लाइव हो जाए।

मैसेज क्लाइंट पेज

क्लाइंट के लिए डायनामिक रूटिंग लागू करने के लिए, हम /pages/user/[username].tsx नामक एक फ़ोल्डर बनाएंगे। यह फ़ोल्डर संरचना हमें प्रत्येक व्यक्तिगत क्लाइंट के लिए उनके उपयोगकर्ता नाम के आधार पर डायनामिक रूट बनाने की अनुमति देगी।

यहां क्लाइंट का मुख्य कंपोनेंट है। यह कंपोनेंट संदेशों की सूची, उपयोगकर्ता नाम आदि के लिए स्टेट रखेगा। हम WebSocket से संदेश, कनेक्शन और डिस्कनेक्शन इवेंट बनाने के लिए useWebSocket हुक का उपयोग कर रहे हैं। जब एक संदेश इवेंट उत्सर्जित होता है, तो संदेश को संदेश सूची में जोड़ा जाएगा और MessageDisplay कंपोनेंट को री-रेंडर किया जाएगा।

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); }; // WebSocket इवेंट संभालना 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 कंपोनेंट हैं:

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> ); };

क्लाइंट को चैट इतिहास प्रदान करने के लिए, हम 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 ऐप अब काम कर रहा है। पेज रीफ्रेश करें, क्लाइंट बनाएं और उनमें से एक पर नेविगेट करें। आप क्लाइंट पेज देखेंगे। लेकिन फिर भी, हमें मैसेज फ्लो को संभालने के लिए मैसेज सर्वर की आवश्यकता है।

मैसेज सर्वर बनाना

सर्वर की संरचना काफी सरल है। हम Node.js, ws लाइब्रेरी, और Upstash Kafka का उपयोग करके इसे काम करने जा रहे हैं। पहले, chat-app folder के अंदर एक server फ़ोल्डर बनाएं।

mkdir server cd server

server फ़ोल्डर के अंदर, हम आवश्यकताएँ इंस्टॉल करेंगे और फ़ाइलों को कॉन्फ़िगर करेंगे।

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

फिर, हम /server/message_server.ts फ़ाइल के अंदर WebSocket, Kafka Producer और Kafka Consumer क्लाइंट बनाने जा रहे हैं:

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>();

WebSocket के साथ बातचीत करने के लिए, हम connection और message इवेंट बना रहे हैं।

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); }); });

अंत में, हम पूर्वनिर्धारित अंतराल के साथ संदेशों का उपभोग करने वाले थ्रेड को बनाएंगे और चलाएंगे:

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)); } }

सब कुछ तैयार है। हमारा ऐप अब आकर्षण की तरह काम करना चाहिए। यदि आप मैसेज सर्वर को स्थानीय रूप से चलाते हैं और क्लाइंट पेज को रीफ्रेश करते हैं, तो आप क्लाइंट्स के बीच प्रसारित संदेशों को देख सकते हैं। नीचे दिए गए कमांड TS फ़ाइल को ट्रांसपाइल करेंगे और सर्वर को localhost:8000 पर चलाएंगे

tsc message_server.ts node message_server.js

डिप्लॉयमेंट

हम डिप्लॉयमेंट के लिए Fly.io का उपयोग करेंगे। यदि आपके पास पहले से नहीं है तो शुरू करने से पहले कृपया एक खाता बनाएं।

मैसेज सर्वर डिप्लॉय करना

server फ़ोल्डर पर जाएं और flyctl CLI टूल इंस्टॉल करें, और शेल के माध्यम से अधिकृत करें

npm install flyctl flyctl auth login

कॉन्फ़िगरेशन फ़ाइलें बनाने के लिए flyctl init चलाएं। यह एक fly.toml बनाएगा। fly.toml पर जाएं और WebSocket कनेक्शन कॉन्फ़िगरेशन के लिए निम्नलिखित लाइनें डालें:

[[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

अब, सर्वर के लिए अंतिम चरण। flyctl deploy चलाएं, और हम जाने के लिए तैयार हैं! जब डिप्लॉयमेंट प्रक्रिया पूरी हो जाती है, तो flyctl आपके सर्वर के लिए एक एंडपॉइंट प्रदान करेगा। कृपया उस एंडपॉइंट को कॉपी करें। हमारे मामले में, एंडपॉइंट message-server.fly.dev है।

Next एप्लिकेशन डिप्लॉय करना

Next.js एप्लिकेशन डिप्लॉय करने से पहले, हमें मैसेज सर्वर के डिप्लॉयमेंट एंडपॉइंट को एम्बेड करने की आवश्यकता है। कृपया pages/user/[username].tsx फ़ाइल में WebSocket URL को ws://localhost:8080 से flyctl से एंडपॉइंट से बदलें, जिसे wss:// उपसर्ग के साथ जोड़ा गया है। हमारे मामले में, यह wss://message-server.fly.dev है।

फिर, next-chat-app फ़ोल्डर में, server के समान कमांड चलाएं। इस बार, हमें fly.toml फ़ाइल संपादित करने की आवश्यकता नहीं है, इसलिए हम उस चरण के बिना आगे बढ़ सकते हैं।

flyctl init flyctl deploy 

हम हो गए! यदि आप flyctl open कमांड चलाते हैं, तो आपको आपके डिप्लॉयड प्रोजेक्ट पर नेविगेट किया जाएगा।

निष्कर्ष और सुझाव

साथ चलने के लिए धन्यवाद!

आप प्रोजेक्ट का Github रिपॉजिटरी यहां पा सकते हैं।

यदि आप प्रोजेक्ट पर काम करना जारी रखना चाहते हैं, तो यहां कुछ सुझाव हैं:

• वर्तमान में, जब भी पेज रीलोड होता है, Upstash Redis में संग्रहीत सभी संदेश फ्लश हो रहे हैं। यह व्यवहार pages/index.tsx फ़ाइल में कोड द्वारा नियंत्रित होता है, विशेष रूप से getServerSideProps फ़ंक्शन के भीतर। हालांकि, एक महत्वपूर्ण मुद्दा तब उत्पन्न होता है जब कोई उपयोगकर्ता पेज को रीलोड करने का फैसला करता है, इसके परिणामस्वरूप चैटरूम में सभी प्रतिभागियों के लिए चैट इतिहास हट जाता है।
  इस मुद्दे को हल करने के लिए, एक अनुशंसित समाधान में हर बार एक संदेश भेजे जाने पर चैट इतिहास के लिए TTL का विस्तार करने का कार्यान्वयन शामिल है। यह सुधार यह सुनिश्चित करेगा कि पेज रीलोड के बाद भी चैट इतिहास सुलभ और संरक्षित रहे।

• आप कई चैटरूम सुविधा लागू कर सकते हैं। इसे प्राप्त करने के लिए, आप प्रत्येक चैटरूम के लिए अद्वितीय नामों के साथ कई Kafka टॉपिक बना सकते हैं। दूसरा तरीका सही डेटा संरचनाओं का उपयोग करके मैसेज सर्वर पर ही इसे संभालना है।

• आप कई मैसेज सर्वर, और सर्वोत्तम सिस्टम डिज़ाइन प्रथाओं को लागू करने के लिए एक लोड बैलेंसर भी लागू कर सकते हैं।

यदि आपके कोई प्रश्न हैं, तो आप मुझसे fahreddin@upstash.com पर संपर्क कर सकते हैं