Node.js تکمیلی؛ رویداد، جریان داده و سرور چت

درخواست دانلود را مثل پرکردن یک لیوان از مخزن تصور کنید: اگر خروجی باریک است، ریختن سریع‌تر فقط صف را بزرگ می‌کند. Stream در Node.js راه کارکردن تدریجی با داده است؛ لازم نیست همیشه تمام فایل یا پاسخ را یک‌جا وارد حافظه کنیم. مسئلهٔ این درس سرعت بیشتر به هر قیمت نیست، هماهنگ‌کردن تولیدکننده و مصرف‌کننده است. در چت هم کاربر کند وجود دارد؛ سرور نباید برای خوشحال نگه‌داشتن یک اتصال، صف نامحدود از پیام‌ها نگه دارد.

مثال‌ها را در پوشه‌ای مستقل با Node.js ۲۴ یا بالاتر اجرا می‌کنیم و هیچ فایل واقعی کاربر را نمی‌خوانیم. Readable منبع خواندن است و Writable مقصد نوشتن؛ قطعهٔ داده یا chunk بخشی است که در یک نوبت منتقل می‌شود. مرز این قطعه لزوماً مرز یک سطر، یک کلمه یا یک پیام کاربردی نیست. همین تمایز کمک می‌کند بعداً قطعهٔ شبکه را به‌اشتباه یک شیء JSON کامل فرض نکنیم. قرارداد پیام WebSocket را کتابخانه جدا از این قطعه‌ها مدیریت می‌کند.

فشار برگشتی؛ درخواست مکث از تولیدکننده

Backpressure یا فشار برگشتی یعنی مصرف‌کننده اعلام می‌کند فعلاً دادهٔ بیشتری تولید نکن تا صف قابل کنترل بماند. در Writable، مقدار false از write معمولاً یعنی قطعه وارد مسیر نوشتن شده، اما برای ادامه باید منتظر ظرفیت بمانیم. این مقدار به‌تنهایی گزارش شکست ذخیره نیست و نباید همان قطعه را دوباره بی‌محابا بنویسید؛ آن کار داده را تکراری می‌کند. خطای واقعی مسیر جدا دارد. نام رویداد drain به اعلام امکان ادامهٔ نوشتن مربوط است، نه تأیید کسب‌وکار برنامه.

highWaterMark آستانهٔ صف داخلی است؛ آن را تضمین قطعی بیشینهٔ حافظهٔ کل برنامه ندانید. اندازهٔ قطعه، لایه‌های دیگر و داده‌هایی که خودتان در آرایه نگه می‌دارید نیز مؤثرند. در حالت بایتی، اندازه با بایت سنجیده می‌شود؛ در objectMode تعداد شیءها مهم است و یک شیء می‌تواند خودش بسیار بزرگ باشد. بنابراین گفتن «فقط ده شیء در صف دارم» بدون محدودیت اندازهٔ هر شیء، سیاست حافظهٔ کاملی نیست. واحد اندازه‌گیری را کنار تنظیمات ثبت کنید.

برای انتقال کامل، pipeline را به کار بگیرید

فایل stream-pipeline.mjs را با کد زیر بسازید و node stream-pipeline.mjs را اجرا کنید. منبع، سه Buffer کوچک می‌سازد و مقصد با تأخیر آموزشی پنج میلی‌ثانیه‌ای آن‌ها را مصرف می‌کند. تأخیر فقط برای ملموس‌شدن مصرف‌کنندهٔ کند است و شبیه‌سازی دقیق دیسک یا شبکه نیست. pipeline نسخهٔ Promise دارد؛ با await پایان موفق یا شکست انتقال را می‌فهمیم. این همان کاربرد عملی #async-await و #promise است: پایان یک عملیات واقعی را دنبال می‌کنیم، نه زمان تقریبی را.

import assert from 'node:assert/strict';
import { Readable, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

async function transfer({ failAt = 0 } = {}) {
  let chunks = 0;
  let bytes = 0;
  const source = Readable.from(
    [Buffer.from('سلام\n'), Buffer.from('Node\n'), Buffer.from('جریان\n')],
    { objectMode: false, highWaterMark: 8 }
  );
  const sink = new Writable({
    highWaterMark: 8,
    write(chunk, encoding, callback) {
      setTimeout(() => {
        chunks++;
        if (chunks === failAt) return callback(new Error('Sink failed'));
        bytes += chunk.length;
        callback();
      }, 5);
    }
  });
  try {
    await pipeline(source, sink);
    return { bytes, chunks };
  } catch (error) {
    assert.equal(source.destroyed, true);
    assert.equal(sink.destroyed, true);
    throw error;
  }
}
const result = await transfer();
assert.equal(result.bytes, Buffer.byteLength('سلام\nNode\nجریان\n'));
console.log(`bytes=${result.bytes}`);
await assert.rejects(transfer({ failAt: 1 }), /Sink failed/);
console.log('failure handled; fresh streams used');

خروجی موفق bytes=25 و سپس failure handled; fresh streams used است. شمار بایت را از Buffer.byteLength متن کامل می‌گیریم، چون تعداد نویسهٔ فارسی برابر اندازهٔ UTF-8 آن نیست. شمار chunk را عمداً معیار محتوای درست قرار نداده‌ایم؛ به مرزهای داخلی قطعه‌ها برای ساخت قرارداد تکیه نکنید. برنامه فقط شمارنده نگه می‌دارد و همهٔ قطعه‌ها را دوباره در آرایه جمع نمی‌کند؛ جمع‌کردن بی‌پایان قطعه‌ها، فایدهٔ مصرف تدریجی را از بین می‌برد.

در اجرای دوم، نخستین نوشتن با callback خطا پایان می‌یابد و pipeline رد می‌شود. آزمون بررسی می‌کند هر دو جریان از بین رفته‌اند و سپس همان خطا را به فراخواننده می‌دهد. برای اجرای بعد، source و sink تازه می‌سازیم. جریانِ شکست‌خورده را مثل یک تابع خالص قابل استفادهٔ دوباره تصور نکنید؛ وضعیت داخلی و شنونده‌های باقی‌مانده می‌توانند مهم باشند. pipeline بسیاری از کارهای هماهنگی و رسیدگی را انجام می‌دهد، ولی مجوز نادیده‌گرفتن عمر منابع یا استفادهٔ بی‌برنامهٔ دوباره از آن‌ها نیست.

در پیاده‌سازی Writable، callback را دقیقاً یک بار برای هر نوشتن صدا بزنید. اگر هم در شاخهٔ خطا و هم انتهای تابع آن را اجرا کنید، قرارداد داخلی جریان را می‌شکنید. در کد ما return پیش از callback خطا همین دو بار اجراشدن را مانع می‌شود. اگر عملیات واقعی یک Promise برمی‌گرداند، موفقیت و شکست آن را به callback وصل کنید و آن Promise را رها نکنید. نبودن پیام خطا روی صفحه، دلیل موفق‌بودن نوشتن نیست؛ مسیر رسیدگی باید قابل آزمون باشد.

false به معنی گم‌شدن قطعه نیست

برای مشاهدهٔ مستقیم این تفاوت، کد بعدی را در stream-write.mjs قرار دهید. آستانهٔ مقصد عمداً یک بایت است، اما قطعه دو بایت دارد. write مقدار false می‌دهد؛ با این حال همان دو بایت مصرف می‌شوند. پس باید پیش از تولید قطعهٔ بعدی مکث کنیم. این مثال مقصدی کنترل‌شده و بدون خطای عمدی دارد؛ برای خط لولهٔ واقعی چند جریان، معمولاً همان pipeline روشن‌تر از دست‌نویس‌کردن تمام حالت‌های drain، پایان، لغو و خطاست.

import assert from 'node:assert/strict';
import { Writable } from 'node:stream';
import { once } from 'node:events';
import { finished } from 'node:stream/promises';

let received = 0;
const sink = new Writable({
  highWaterMark: 1,
  write(chunk, encoding, callback) {
    setTimeout(() => { received += chunk.length; callback(); }, 5);
  }
});
const completion = finished(sink, { cleanup: true });
const accepted = sink.write(Buffer.from('AB'));
assert.equal(accepted, false);
await once(sink, 'drain');
sink.end();
await completion;
assert.equal(received, 2);
console.log(`write returned ${accepted}; received=${received}`);

انتظار write returned false; received=2 است. finished از ابتدا پایان مقصد را دنبال می‌کند و پس از end منتظر آن می‌مانیم. once برای drain نیز در صورت رخ‌دادن error می‌تواند رد شود؛ حذف مسیر خطا از کد تولیدی راه حل نیست. اگر در برنامهٔ واقعی نتیجهٔ انتقال را به کاربر نشان می‌دهید، موفقیت را پس از پایان معتبر اعلام کنید. افزودن یک setTimeout و نوشتن «انجام شد» بعد از آن، اندازه‌گیری پایان کار محسوب نمی‌شود.

متن فارسی ممکن است میان دو قطعه شکسته شود

فرض کنید بایت‌های یک حرف فارسی در دو نوبت برسند. تبدیل مستقل هر قطعه به متن ممکن است علامت جایگزین بسازد، چون قطعهٔ اول هنوز کد کامل آن حرف را ندارد. StringDecoder بخشی از کتابخانهٔ داخلی Node.js است که بایت ناقص انتهای قطعه را برای ادامه نگه می‌دارد. این مسئله دربارهٔ مرز رمزگذاری متن است؛ تجزیهٔ JSON، محدودیت طول پیام و اعتبار داده وظایف دیگری هستند. در واژه‌نامهٔ #string و #type-conversion تفاوت متن و تبدیل نوع را مرور کنید.

کد زیر را در stream-utf8.mjs اجرا کنید. شکست قطعه‌ها عمدی است: نخست فقط یک بایت داده‌ایم. خروجی باید دقیقاً سلام باشد و assert همین را بررسی می‌کند. end پایان ورودی را اعلام می‌کند تا باقی‌مانده نیز پردازش شود. این نمونه فقط تبدیل متن را نشان می‌دهد؛ برای خواندن فایل بزرگ لازم است علاوه بر تبدیل درست، مجموع مصرف حافظه و طول رکوردها هم محدود شود. یک سطر بدون پایان می‌تواند حتی با جریان‌خوانی به مشکل دیگری تبدیل شود.

import assert from 'node:assert/strict';
import { StringDecoder } from 'node:string_decoder';

const bytes = Buffer.from('سلام', 'utf8');
const parts = [bytes.subarray(0, 1), bytes.subarray(1, 5), bytes.subarray(5)];
const decoder = new StringDecoder('utf8');
let text = '';
for (const part of parts) text += decoder.write(part);
text += decoder.end();
assert.equal(text, 'سلام');
console.log(text);

این مفهوم در چت چه کاربردی دارد؟

کتابخانهٔ ws پیام WebSocket را در اختیار سرور می‌گذارد، اما کندبودن مقصد همچنان واقعی است. پروژهٔ ما از bufferedAmount و اندازهٔ بستهٔ بعدی برای کنترل صف ارسال استفاده می‌کند و اگر اتصال بیش از حد کند باشد آن را قطع می‌کند. این سیاست با drain یک Writable عیناً یکی نیست. API بومی WebSocket مرورگر نیز فشار برگشتی خودکار برای مصرف تمام پیام‌های دریافتی در اختیار شما نمی‌گذارد؛ تعداد پیام نمایشی و نرخ ارسال را باید در سطح برنامه کنترل کنید.

در این پروژه سقف تاریخچه پنجاه پیام است و نگه‌داشتن آن تصمیمی جدا از صف شبکه است. اگر تاریخچه محدود باشد ولی برای هر کلاینت صف بی‌پایان بسازیم، مشکل حافظه حل نشده است. برعکس، قطع اتصال کند نیز تضمین نمی‌کند آخرین پیام به کاربر رسیده باشد. در درس اتصال سالم، این تفاوت را با وضعیت نامعلوم ارسال و تأیید کاربردی پیوند می‌دهیم. هدف، روشن‌بودن قرارداد است تا رابط وعده‌ای فراتر از اطلاعات واقعی ندهد.

تمرین، پاسخ و معیار بررسی

تأخیر مصرف‌کنندهٔ نمونه را از پنج به بیست میلی‌ثانیه تغییر دهید و سه بار اجرا کنید. پاسخ درست این است که تعداد بایت محتوا همچنان ۲۵ باشد؛ از مدت اجرای یک نمونهٔ کوچک نتیجهٔ عمومی دربارهٔ کارایی نگیرید. سپس failAt را یک قرار دهید و بررسی کنید هیچ برچسب موفقیتی برای انتقال شکست‌خورده چاپ نشود. پاسخ راهنما: چاپ موفقیت باید پس از await pipeline باشد و رسیدگی خطا در فراخواننده انجام شود، نه در یک تایمر مستقل.

در تمرین دوم، متن را به «آموزش فارسی» تغییر دهید و آن را از وسط بایت یک حرف بشکنید. انتظار را با Buffer.byteLength بسنجید و متن نهایی را با assert مقایسه کنید. معیار تحویل شامل حفظ محتوا، پایان قابل مشاهده، رسیدگی شکست و ساخت جریان تازه برای کار بعدی است. اگر با افزایش highWaterMark فقط خطا را دیرتر می‌بینید، علت کندی یا صف نامحدود هنوز بررسی نشده است. توضیح دهید هر محدودیت مربوط به بایت، شیء یا تعداد پیام است.

برای پیوند با درس‌های قبلی، بخش UTF-8 و فایل در Node.js و درس ساخت API را مرور کنید؛ اینجا به جای بازگویی مقدمات فایل، نحوهٔ عبور تدریجی و محدود داده را دنبال کردیم.

UTF-8 و فایل‌ها: https://www.darsavand.com/courses/nodejs-windows-backend-start?lesson=08d325f8-3c97-4498-9f94-984029361479
بدنهٔ درخواست و API: https://www.darsavand.com/courses/nodejs-http-server-page-api-tls?lesson=7a83fd1a-3584-409c-8eb4-dd51e21144ce

منابع رسمی و مطالعهٔ تکمیلی

Node.js Stream API: https://nodejs.org/api/stream.html
Node.js Learn، Backpressure: https://nodejs.org/en/learn/modules/backpressuring-in-streams
Node.js StringDecoder: https://nodejs.org/api/string_decoder.html
MDN، محدودیت WebSocket: https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API

تألیف و مثال‌های اختصاصی: تحریریه درس‌آوند.

تألیف و مثال‌های اختصاصی: تحریریه درس‌آوند. فایل‌های کامل پروژه و راهنمای اجرا و دیباگ:
https://www.darsavand.com/resources/simple-websocket-chat-project