การแจ้งเตือนแบบเรียลไทม์ด้วย Upstash Redis, การดำเนินการของเซิร์ฟเวอร์ Next.js และ Vercel:คำแนะนำทีละขั้นตอน

การตั้งค่าโครงการ

หากต้องการตั้งค่า เพียงโคลน repo ของแอปแล้วทำตามบทช่วยสอนนี้เพื่อเรียนรู้ทุกอย่างที่อยู่ในนั้น หากต้องการแยกโครงการ ให้รัน:

git clone https://github.com/rishi-raj-jain/upstash-nextjs-publish-messages-with-sse-example
cd upstash-nextjs-publish-messages-with-sse-example
npm install

เมื่อคุณโคลน repo แล้ว คุณจะต้องสร้างไฟล์ .env คุณกำลังจะเพิ่มรายการที่เราบันทึกไว้จากส่วนด้านบน

มันควรมีลักษณะดังนี้:

# .env
 
# Obtained from the steps as above
 
# Upstash Redis Secrets
UPSTASH_REDIS_URL="rediss://default:...@...-...-...-....upstash.io:..."
UPSTASH_REDIS_REST_URL="https://...-...-...-....upstash.io"
UPSTASH_REDIS_REST_TOKEN="...="

โปรดสังเกตว่าที่ 47 ตัวแปรระบุว่า "rediss" ไม่ใช่ "redis" นั่นคือการใช้ตัวเลือก TLS/SSL

หลังจากขั้นตอนเหล่านี้ คุณควรจะสามารถเริ่มต้นสภาพแวดล้อมภายในเครื่องได้โดยใช้คำสั่งต่อไปนี้:

npm run dev

โครงสร้างพื้นที่เก็บข้อมูล

นี่คือโครงสร้างโฟลเดอร์หลักของโครงการ ฉันได้ทำเครื่องหมายสีแดงในไฟล์ที่จะกล่าวถึงเพิ่มเติมในโพสต์นี้ที่เกี่ยวข้องกับสิ่งต่อไปนี้:

• ทำความเข้าใจช่องข้อความใน Upstash Redis
• การสร้าง API เหตุการณ์ที่เซิร์ฟเวอร์ส่งในเราเตอร์แอป Next.js
• ตั้งค่าการดำเนินการเซิร์ฟเวอร์ Next.js ด้วย Upstash Redis เพื่อเผยแพร่การแจ้งเตือน
• ตั้งค่าส่วนหน้า Next.js เพื่อฟังและแสดงการแจ้งเตือนแบบเรียลไทม์อย่างต่อเนื่อง

ทำความเข้าใจช่องข้อความใน Upstash Redis

ใน Upstash Redis โมเดลการเผยแพร่/สมัครสมาชิกจะเป็นแกนหลักของช่องทางข้อความ ผู้เผยแพร่เผยแพร่ข้อความไปยังช่องที่ระบุชื่อ และสมาชิกสามารถรับข้อความจากช่องใดช่องหนึ่งได้แบบเรียลไทม์ โมเดลนี้ช่วยให้สามารถสื่อสารได้อย่างราบรื่นระหว่างส่วนต่างๆ ของแอปพลิเคชัน

ต่อไปนี้คือวิธีการเผยแพร่ข้อความไปยังช่องโดยใช้ไลบรารีที่เข้ากันได้กับ Edge 52 👇🏻

import { Redis } from '@upstash/redis'
 
// Connect to an Upstash Redis instance
const redis = Redis.fromEnv()
 
// Publish a message to the Upstash Redis instance
await redis.publish('posts', JSON.stringify({ date: new Date().toString(), message: "I am a new message." }))

ต่อไปนี้คือวิธีที่สมาชิกสามารถฟังช่อง Upstash Redis (ที่นี่ 60 ) พร้อมด้วยไลบรารีที่เข้ากันได้กับโหนด 79 👇🏻

// Use ioredis to be able to subscribe to an Upstash Redis instance
import Redis from 'ioredis'
 
// Create an Upstash Redis Subscriber instance
const redisSubscriber = new Redis(process.env.UPSTASH_REDIS_URL)
 
// Define the key to listen and publish messages to
const setKey = 'posts'
 
// Subscribe to Redis updates for the key: "posts"
// In case of any error, just log it
redisSubscriber.subscribe(setKey, (err) => {
 if (err) console.log(err)
})
 
// Listen for new posts from Redis
redisSubscriber.on('message', (channel, message) => {
 // Log the data when the channel message is received is same as the message is published to
 if (channel === setKey) console.log(mesage)
})

เมื่อเผยแพร่ข้อความไปยังช่อง สมาชิกทุกคนจะได้รับข้อความทันที ทำให้สามารถสื่อสารได้อย่างมีประสิทธิภาพและเรียลไทม์ภายใน Upstash Redis

การสร้าง API เหตุการณ์ที่เซิร์ฟเวอร์ส่งในเราเตอร์แอป Next.js

เหตุการณ์ที่เซิร์ฟเวอร์ส่งเป็นวิธีที่มีประสิทธิภาพในการส่งข้อมูลใหม่แบบเรียลไทม์โดยไม่จำเป็นต้องร้องขอจากไคลเอ็นต์หลายรายการ ต่างจากกลไกการตอบกลับคำขอแบบดั้งเดิม SSE ช่วยให้สามารถส่งข้อมูลในทิศทางเดียวจากเซิร์ฟเวอร์ไปยังไคลเอนต์ผ่านการเชื่อมต่อ HTTP เดียวที่มีอายุการใช้งานยาวนาน

ต่อไปนี้คือวิธีที่คุณสามารถใช้เหตุการณ์ที่เซิร์ฟเวอร์ส่งในเราเตอร์แอป Next.js 👇🏻

// File: app/api/stream/route.js
 
// Prevents this route's response from being cached on Vercel
export const dynamic = 'force-dynamic'
 
export async function GET() {
 const encoder = new TextEncoder()
 // Create a streaming response
 const customReadable = new ReadableStream({
 start(controller) {
 const message = 'Hey, I am a message.'
 controller.enqueue(encoder.encode(`data: ${message}\n\n`))
 },
 })
 // Return the stream response and keep the connection alive
 return new Response(customReadable, {
 // Set the headers for Server-Sent Events (SSE)
 headers: {
 'Content-Type': 'text/event-stream; charset=utf-8',
 Connection: 'keep-alive',
 'Cache-Control': 'no-cache, no-transform',
 'Content-Encoding': 'none'
 },
 })
}

การเผยแพร่ข้อความไปยัง Upstash Redis โดยใช้การดำเนินการของเซิร์ฟเวอร์ Next.js

การดำเนินการของเซิร์ฟเวอร์ Next.js ช่วยให้คุณสามารถกำหนดตรรกะฝั่งเซิร์ฟเวอร์ได้โดยตรงภายในโค้ดส่วนหน้าใน Next.js ซึ่งจะช่วยประหยัดกระบวนการสร้างเส้นทาง API ด้วยตนเอง และลดความยุ่งยากในการส่งและติดตามสถานะการส่งแบบฟอร์ม

ด้วย 88 ที่ด้านบนของฟังก์ชัน เราสามารถตรวจสอบให้แน่ใจว่าฟังก์ชันต่างๆ ทำงานเฉพาะฝั่งเซิร์ฟเวอร์เท่านั้น ภายในการดำเนินการเซิร์ฟเวอร์การส่งแบบฟอร์มของเรา เราจะแยก 92 ค่าจากแบบฟอร์ม ใช้ Vercel Header เพื่อรับประเทศของผู้ใช้และเผยแพร่ข้อมูลเป็นข้อความไปที่ 101 ช่องทางข้อความ Upstash Redis

// File: app/actions.jsx
 
'use server'
 
import { Redis } from '@upstash/redis'
import { headers } from 'next/headers'
 
// The function that takes care of obtaining the country code from Vercel headers
// And publishing messages to the Upstash Redis database with the current timestamp
export async function publishNotification(formData) {
 'use server'
 const redis = Redis.fromEnv()
 
 // Extract the message in the form submitted
 const message = formData.get('message')
 
 // Obtain country of the user using Vercel's x-vercel-ip-country header
 const headersList = headers()
 const country = headersList.get('x-vercel-ip-country')
 
 // Publish the message to the "posts" channel in Upstash Redis
 await redis.publish(
 'posts',
 JSON.stringify({
 message,
 country,
 date: new Date().toString(),
 }))
}

หากต้องการเรียกใช้ Server Action นี้เมื่อมีการส่งแบบฟอร์ม เราจะส่งผ่านเป็นตัวจัดการไปยังแบบฟอร์ม 115 เหตุการณ์

// File: app/page.jsx
 
// Import the server action
import { publishNotification } from './actions'
 
const Home = () => {
 return (
 <>
 <div>
 <form
 
 /* set the server action to invoked as form is submitted */
 action={publishNotification}
 
 >
 {/* Place a client side form component here */}
 </form>
 </div>
 </>
 )
}
 
export default Home

ตั้งค่าส่วนหน้า Next.js เพื่อแสดงสถานะรอดำเนินการระหว่างการส่งแบบฟอร์มด้วย hook useFormStatus ของ React

รหัสต่อไปนี้สาธิตการตั้งค่าส่วนประกอบฝั่งไคลเอ็นต์แบบฟอร์ม Next.js เพื่อจัดการการส่งแบบฟอร์มและแสดง 120 สถานะโดยใช้ 134 ของ React ตะขอ มาแจกแจงองค์ประกอบสำคัญของโค้ด:

• การนำเข้า 145 ที่เพิ่งเปิดตัว hook จาก React](https://react.dev/reference/react-dom/hooks/useFormStatus) ที่ให้ข้อมูลสถานะของการส่งแบบฟอร์มครั้งล่าสุดแก่คุณ
• การใช้ 155 ตัวแปรปฏิกิริยาสถานะที่ระบุว่ากำลังส่งแบบฟอร์มอยู่หรือไม่
• หากการส่งไม่อยู่ระหว่างดำเนินการ 167 แบบฟอร์ม
• ใช้ 178 บูลีนเพื่อแสดงสถานะตามเงื่อนไขของแบบฟอร์ม

'use client'
 
import { useEffect } from 'react'
import { useFormStatus } from 'react-dom'
 
const Form = () => {
 
 // Use React's useFormStatus hook to detect form submission state
 const { pending } = useFormStatus()
 
 useEffect(() => {
 // If the form is not pending, reset the form
 if (!pending) document.getElementById('publish-form').reset()
 }, [pending])
 
 return (
 <>
 <input placeholder="Your message" className="border rounded px-3 outline-none focus:border-black/50 py-2" type="text" name="message" required />
 <button
 /* Disable button click while the form submission is pending */
 disabled={pending}
 className="hover:border-black/50 max-w-max border rounded py-1 px-3" type="submit"
 >
 {/* Display "pending" state placeholder */}
 {pending ? (
 <div className="flex flex-row gap-x-2 items-center">
 <div className="animate-spin border border-gray-800 rounded-full h-[15px] w-[15px]"></div>
 <span>Publishing</span>
 </div>
 ) : (
 <>Publish Notification &rarr;</>
 )}
 </button>
 </>
 )
}
 
export default Form

ตั้งค่าส่วนหน้า Next.js เพื่อฟังเหตุการณ์ที่เซิร์ฟเวอร์ส่งอย่างต่อเนื่อง

ในส่วนนี้ เราจะได้เรียนรู้วิธีตั้งค่า Listener ขั้นต่ำสำหรับข้อความ Server-Sent Events API และวิธีการคงการเชื่อมต่อกับ SSE API

การฟัง API เหตุการณ์ที่เซิร์ฟเวอร์ส่งใน React Frontend

หากต้องการฟัง SSE API ในส่วนประกอบฝั่งไคลเอ็นต์ของเราใน React เราใช้ 180 ตะขอ หากต้องการสร้างการเชื่อมต่อกับ SSE API ให้สร้าง 195 ใหม่ อินสแตนซ์ชี้ไปที่ 205 จุดสิ้นสุด จากนั้นแนบตัวฟังเหตุการณ์สำหรับ 217 เหตุการณ์ที่ข้อมูลขาเข้าจากสตรีมถูกแยกวิเคราะห์เป็น JSON และประมวลผลเพิ่มเติมหรือแสดงในส่วนประกอบ

สุดท้ายนี้ โค้ดจะมีฟังก์ชันการล้างข้อมูลเพื่อปิดการเชื่อมต่อ SSE เมื่อถอดส่วนประกอบออก เพื่อป้องกันหน่วยความจำรั่วที่อาจเกิดขึ้น

import { useEffect, useState } from 'react'
 
const ClientSideComponent = () => {
 
 useEffect(() => {
 
 // Initiate the first call to connect to SSE API
 const eventSource = new EventSource('/api/stream')
 
 eventSource.addEventListener('message', (event) => {
 // Parse the data received from the stream into JSON
 // Add it the list of messages seen on the page
 const tmp = JSON.parse(event.data)
 // Do something with the obtained message
 })
 
 // As the component unmounts, close listener to SSE API
 return () => {
 eventSource.close()
 }
 
 }, [])
 
 return <></>
}
 
export default ClientSideComponent

การเชื่อมต่อกับ API เหตุการณ์ที่เซิร์ฟเวอร์ส่งอย่างต่อเนื่องใน React Frontend

ในเวอร์ชันปรับปรุงของคอมโพเนนต์ React เราได้ใช้กลไกเพื่อให้แน่ใจว่ามีการเชื่อมต่อกับ SSE API อย่างต่อเนื่องและต่อเนื่อง โดยจัดการข้อผิดพลาดและเชื่อมต่อใหม่โดยอัตโนมัติ

ซึ่งสามารถทำได้ผ่านทาง 228 ฟังก์ชันที่รับผิดชอบในการสร้างและรักษาการเชื่อมต่อกับ SSE API

นี่คือรายละเอียดฟังก์ชันต่างๆ 👇🏻

1. การเชื่อมต่อเริ่มต้นและการจัดการข้อความ:

ฟังก์ชั่นสร้าง 237 ใหม่ อินสแตนซ์เชื่อมต่อกับ 243 จุดสิ้นสุด

// Function to take care of initial connect to the SSE API
// Also, it reconnects to the SSE API as soon as it shuts down
// This keeps the connection alive - forever with micro second delays
const connectToStream = () => {
 // Connect to /api/stream as the SSE API source
 const eventSource = new EventSource('/api/stream')
 // ..
}

นอกจากนี้ยังตั้งค่าตัวฟังเหตุการณ์สำหรับ 256 เหตุการณ์ที่ข้อมูลขาเข้าจากสตรีมถูกแยกวิเคราะห์เป็น JSON ข้อมูลที่แยกวิเคราะห์แล้วสามารถประมวลผลหรือแสดงในองค์ประกอบ React ได้

const connectToStream = () => {
 // ...
 eventSource.addEventListener('message', (event) => {
 // Parse the data received from the stream into JSON
 // Add it the list of messages seen on the page
 const tmp = JSON.parse(event.data)
 setPosts((prevPosts) => [...prevPosts, tmp])
 })
 // ...
}

1. การจัดการข้อผิดพลาดและการเชื่อมต่อใหม่อัตโนมัติ:

Listener เหตุการณ์เพิ่มเติมถูกตั้งค่าสำหรับ 260 เหตุการณ์ ในกรณีที่มีข้อผิดพลาดใดๆ เช่น การเชื่อมต่อล้มเหลว แหล่งที่มาของเหตุการณ์จะถูกปิด

หลังจากปิด ฟังก์ชันจะใช้ 274 เพื่อเริ่มการเชื่อมต่อใหม่หลังจากล่าช้าอย่างน้อย 1 มิลลิวินาที ความล่าช้าเล็กน้อยนี้ช่วยในการสร้างกระบวนการเชื่อมต่อใหม่ที่ราบรื่นและต่อเนื่องโดยไม่ทำให้เซิร์ฟเวอร์ล้นหลามด้วยการพยายามเชื่อมต่อที่รวดเร็ว

// In case of any error, close the event source
// So that it attempts to connect again
eventSource.addEventListener('error', () => {
 eventSource.close()
 setTimeout(connectToStream, 1)
})

1. การจัดการการปิดแหล่งที่มาของ SSE API:

287 เหตุการณ์ใช้เพื่อตรวจจับเมื่อแหล่ง SSE API ถูกปิด เมื่อปิด ฟังก์ชันจะกำหนดเวลาความพยายามเชื่อมต่อกับสตรีมอีกครั้งหลังจากเกิดความล่าช้าเล็กน้อย

// As soon as SSE API source is closed, attempt to reconnect
eventSource.onclose = () => {
 setTimeout(connectToStream, 1)
}