अपस्टैश काफ्का, रेडिस और नेक्स्ट.जेएस के साथ एक रियल-टाइम चैट ऐप बनाएं

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

डेमो

आरंभ करना

1. अपस्टैश रेडिस डेटाबेस बनाना
2. अपस्टैश काफ्का क्लस्टर बनाना
3. अगला एप्लिकेशन (फ्रंटएंड) बनाना।
4. WebSocket संदेश सर्वर बनाना।
5. एप्लिकेशन को Fly.io पर परिनियोजित करना

अपस्टैश रेडिस डेटाबेस बनाना

अपस्टैश काफ्का क्लस्टर बनाना

अगला ऐप बनाना

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

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

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

ग्राहक पंजीकरण पृष्ठ

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

संदेश क्लाइंट पृष्ठ

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);
 };
 
 // handling WebSocket events
 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>
 );
}

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

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

संदेश सर्वर बनाना

mkdir server
cd server

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

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

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

tsc message_server.ts
node message_server.js

तैनाती

संदेश सर्वर परिनियोजन

npm install flyctl
flyctl auth login

[[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 init
flyctl deploy

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

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

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

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