هناك عدة أنواع من التدفقات في العقدة

هناك 4 أنواع من تدفقات العقدة: 1. قابل للقراءة (دفق قابل للقراءة). يجب تنفيذ طريقة "_read" لإرجاع المحتوى؛ 2. قابل للكتابة (دفق قابل للكتابة)، يجب تنفيذ طريقة "_write" لقبول المحتوى 3. مزدوج (دفق قابل للقراءة والكتابة)، و"_read" و" يجب تنفيذ طرق "_write" لقبول المحتوى وإعادته؛ 4. التحويل (دفق التحويل)، تحتاج إلى تنفيذ طريقة "_transform" لتحويل المحتوى المستلم وإرجاع المحتوى.

بيئة تشغيل هذا البرنامج التعليمي: نظام Windows 7، إصدار Nodejs 16، كمبيوتر DELL G3.

يعد Stream مفهومًا أساسيًا جدًا في Nodejs. يتم تنفيذ العديد من الوحدات الأساسية بناءً على التدفقات وتلعب دورًا مهمًا للغاية. في الوقت نفسه، يعد التدفق أيضًا مفهومًا صعب الفهم، ويرجع ذلك أساسًا إلى عدم وجود الوثائق ذات الصلة، بالنسبة للمبتدئين في NodeJ، غالبًا ما يستغرق الأمر الكثير من الوقت لفهم التدفق قبل أن يتمكنوا من إتقان هذا المفهوم حقًا. بالنسبة لمعظم NodeJs، يتم استخدامه فقط لتطوير تطبيقات الويب، ولا يؤثر الفهم غير الكافي للتدفقات على استخدامها. ومع ذلك، يمكن أن يؤدي فهم التدفقات إلى فهم أفضل للوحدات النمطية الأخرى في NodeJs، وفي بعض الحالات، سيكون لاستخدام التدفقات لمعالجة البيانات نتائج أفضل.

تدفق

Stream عبارة عن واجهة مجردة لمعالجة بيانات التدفق في Node.js. الدفق ليس واجهة فعلية، ولكنه مصطلح عام لجميع الدفق. تتضمن الواجهات الفعلية ReadableStream، وWritableStream، وReadWriteStream.

واجهة ReadableStream يمتد EventEmitter { readable: boolean؛ read(size?: number): string |.setEncoding(encoding: BufferEncoding): this; T Extends WritableStream>(destination: T, options?: { end?: boolean | undefense; }): T; unpipe(destination?: WritableStream): this; unshift(chunk: string | Uint8Array, encoding?: BufferEncoding): void ; Wrap(oldStream: ReadableStream): this; خطأ |.null) => void): boolean; write(str: string, encoding?: BufferEncoding, cb?: (err?: Error | null) => void): boolean; ): this; end(data: string | Uint8Array, cb?: () => void): this; end(str: string, encoding?: BufferEncoding, cb?: () => void): this;}interface ReadWriteStream يمتد ReadableStream، WritableStream { }

يمكن ملاحظة أن ReadableStream وWritableStream كلاهما واجهات ترث فئة EventEmitter (يمكن للواجهات في ts أن ترث الفئات، لأنها مجرد أنواع مدمجة).

فئات التنفيذ المقابلة للواجهات المذكورة أعلاه هي قابلة للقراءة والكتابة والطباعة على الوجهين على التوالي.

هناك 4 أنواع من التدفقات في NodeJs:

تيار قابل للقراءة (يطبق ReadableStream)

دفق قابل للكتابة (ينفذ WritableStream)

دوبلكس هو تيار قابل للقراءة والكتابة (تنفيذ WritableStream بعد وراثة Readable)

تحويل دفق التحويل (موروث من دوبلكس)

لديهم جميعا طرق للتنفيذ:

يحتاج Readable إلى تنفيذ طريقة _read لإرجاع المحتوى

يحتاج Writable إلى تنفيذ طريقة _write لقبول المحتوى

يحتاج دوبلكس إلى تنفيذ طريقتي _read و_write لقبول المحتوى وإرجاعه

يحتاج التحويل إلى تنفيذ طريقة _transform لتحويل المحتوى المستلم وإعادته

قابلة للقراءة

Readable هو نوع من الدفق وله وضعان وثلاث حالات.

وضعين للقراءة:

وضع التدفق: ستتم قراءة البيانات وكتابتها من النظام الأساسي إلى المخزن المؤقت. عندما يمتلئ المخزن المؤقت، سيتم تمرير البيانات تلقائيًا إلى معالج الأحداث المسجل في أسرع وقت ممكن من خلال EventEmitter.

وضع الإيقاف المؤقت: في هذا الوضع، لن يتم تشغيل EventEmitter بشكل فعال لإرسال البيانات، ويجب استدعاء الأسلوب Readable.read() بشكل صريح لقراءة البيانات من المخزن المؤقت للقراءة، مما سيؤدي إلى استجابة لحدث EventEmitter.

ثلاث ولايات:

readableFlowing === null (الحالة الأولية)

readableFlowing === خطأ (وضع الإيقاف المؤقت)

readableFlowing === صحيح (وضع التدفق)

يكون readable.readableFlowing للمجرى خاليًا في البداية.

يصبح صحيحًا بعد إضافة حدث البيانات. عند استدعاء Pause()‎ أو unpipe()‎ أو استقبال ضغط رجعي أو إضافة حدث قابل للقراءة، سيتم ضبط readableFlowing على false. في هذه الحالة، لن يؤدي ربط المستمع بحدث البيانات إلى تحويل readableFlowing إلى true.

يمكن أن يؤدي استدعاء السيرة الذاتية () إلى تحويل التدفق القابل للقراءة للتدفق القابل للقراءة إلى صحيح.

إن إزالة جميع الأحداث القابلة للقراءة هي الطريقة الوحيدة لجعل readableFlowing فارغًا.

يتم تشغيل وصف اسم الحدث القابل للقراءة عندما تكون هناك بيانات جديدة قابلة للقراءة في المخزن المؤقت (سيتم تشغيله في كل مرة يتم فيها إدراج عقدة في تجمع ذاكرة التخزين المؤقت) وسيتم تشغيل البيانات في كل مرة يتم فيها استهلاك البيانات يتم تشغيل دفق الأخطاء عند إغلاق دفق الإغلاق. عند حدوث خطأ، يشير اسم طريقة التشغيل إلى أن القراءة (الحجم) تستهلك بيانات بطول الحجم، مما يشير إلى أن البيانات الحالية أقل من الحجم يتم إرجاع البيانات المستهلكة هذه المرة. عندما لا يتم تمرير الحجم، فهذا يعني استهلاك جميع البيانات الموجودة في تجمع ذاكرة التخزين المؤقت const fs = require('fs'); const readStreams = fs.createReadStream('./EventEmitter.js', { HighWaterMark: 100// تعويم تجمع ذاكرة التخزين المؤقت value})readStreams.on('readable', () => { console.log('buffer full') readStreams.read()// استهلك جميع البيانات الموجودة في تجمع المخزن المؤقت، وأرجع النتيجة وقم بتشغيل حدث البيانات}) readStreams.on('data', (data) => { console.log('data')})

https://github1s.com/nodejs/node/blob/v16.14.0/lib/internal/streams/readable.js#L527

عندما يكون الحجم 0، سيتم تشغيل الحدث القابل للقراءة.

عندما يصل طول البيانات في تجمع ذاكرة التخزين المؤقت إلى القيمة العائمة HighWaterMark، فلن يطلب بيانات الإنتاج بشكل نشط، ولكنه سينتظر حتى يتم استهلاك البيانات قبل إنتاج البيانات.

إذا كان الدفق في حالة الإيقاف المؤقت لا يستدعي القراءة لاستهلاك البيانات، فلن يتم تشغيل البيانات القابلة للقراءة لاحقًا عند استدعاء القراءة للاستهلاك، فسيحدد أولاً ما إذا كان طول البيانات المتبقية بعد هذا الاستهلاك أقل من التعويم وإذا كانت أقل من القيمة العائمة، فسيتم طلب بيانات الإنتاج قبل الاستهلاك. بهذه الطريقة، بعد اكتمال تنفيذ المنطق بعد القراءة، من المرجح أن يتم إنتاج البيانات الجديدة، ثم سيتم تشغيلها مرة أخرى قابلة للقراءة السبب وراء سرعة دفق ذاكرة التخزين المؤقت.

هناك حالتان من التدفق في الحالة المتدفقة

عندما تكون سرعة الإنتاج أبطأ من سرعة الاستهلاك: في هذه الحالة، لن تكون هناك بيانات متبقية بشكل عام في تجمع ذاكرة التخزين المؤقت بعد كل بيانات إنتاج، ويمكن تمرير البيانات المنتجة هذه المرة مباشرة إلى حدث البيانات (لأنه لا أدخل إلى تجمع ذاكرة التخزين المؤقت، لذلك ليست هناك حاجة أيضًا إلى استدعاء القراءة للاستهلاك)، ثم ابدأ على الفور في إنتاج بيانات جديدة، ولن يتم إنتاج البيانات الجديدة حتى يتم تشغيل البيانات الأخيرة مرة أخرى حتى انتهاء الدفق . عندما تكون سرعة الإنتاج أسرع من سرعة الاستهلاك: في هذا الوقت، بعد كل إنتاج للبيانات، عادة ما تكون هناك بيانات غير مستهلكة في تجمع ذاكرة التخزين المؤقت. في هذه الحالة، سيبدأ الاستهلاك التالي للبيانات بشكل عام عند استهلاك البيانات، وبعد ذلك يتم استهلاك البيانات القديمة، ويتم إنتاج بيانات جديدة ووضعها في تجمع ذاكرة التخزين المؤقت

والفرق الوحيد بينهما هو ما إذا كانت البيانات لا تزال موجودة في تجمع ذاكرة التخزين المؤقت بعد إنتاج البيانات. إذا كانت البيانات موجودة، فسيتم دفع البيانات المنتجة إلى تجمع ذاكرة التخزين المؤقت لانتظار الاستهلاك يتم تسليمها مباشرة إلى البيانات دون إضافتها إلى تجمع ذاكرة التخزين المؤقت.

تجدر الإشارة إلى أنه عندما يدخل دفق يحتوي على بيانات في تجمع ذاكرة التخزين المؤقت إلى وضع التدفق من وضع الإيقاف المؤقت، سيتم استدعاء القراءة في حلقة لاستهلاك البيانات حتى يتم إرجاع قيمة فارغة.

وضع الإيقاف المؤقت

في وضع الإيقاف المؤقت، عندما يتم إنشاء دفق قابل للقراءة، يكون الوضع هو وضع الإيقاف المؤقت. بعد الإنشاء، يتم استدعاء الأسلوب _read تلقائيًا لدفع البيانات من مصدر البيانات إلى تجمع المخزن المؤقت حتى تصل البيانات الموجودة في تجمع المخزن المؤقت إلى القيمة العائمة. عندما تصل البيانات إلى القيمة العائمة، سيقوم الدفق القابل للقراءة بتشغيل حدث "قابل للقراءة" لإخبار المستهلك أن البيانات جاهزة ويمكن الاستمرار في استهلاكها.

بشكل عام، يشير الحدث 'readable' إلى نشاط جديد في الدفق: إما أن هناك بيانات جديدة، أو تم الوصول إلى نهاية الدفق. لذلك، قبل قراءة البيانات الموجودة في مصدر البيانات، سيتم أيضًا تشغيل الحدث 'readable'؛

في وظيفة المعالج للحدث "القابل للقراءة" للمستهلك، يتم استهلاك البيانات الموجودة في تجمع المخزن المؤقت بشكل نشط من خلال Stream.read(size).

const { Readable } = require('stream')let count = 1000const myReadable = new Readable({ HighWaterMark: 300, // سيتم استخدام طريقة قراءة المعلمة كطريقة _read للتيار للحصول على البيانات المصدر المقروءة( size) { // افترض أن بيانات المصدر لدينا تحتوي على 1000 1 ثانية Letchunk = null // عملية قراءة البيانات غير متزامنة بشكل عام، مثل عملية الإدخال والإخراج setTimeout(() => { if (count > 0) { LetchunkLength = Math .min( count, size)chunk = '1'.repeat(chunkLength) count -=chunkLength } this.push(chunk) }, 500) }})//readablemyReadable.on(' سيتم تشغيله في كل مرة يتم فيها تشغيل البيانات تم الدفع بنجاح إلى تجمع ذاكرة التخزين المؤقت) readable', () => { const Chunk = myReadable.read()// استهلك جميع البيانات الموجودة في تجمع ذاكرة التخزين المؤقت الحالي console.log(chunk.toString())})

تجدر الإشارة إلى أنه إذا كان حجم القراءة (الحجم) أكبر من قيمة التعويم، فسيتم إعادة حساب قيمة التعويم الجديدة، وتكون قيمة التعويم الجديدة هي القوة الثانية التالية للحجم (الحجم <= 2^n، n يأخذ الحد الأدنى للقيمة)

// لن يكون حجم hwm أكبر من 1 جيجابايت.const MAX_HWM = 0x40000000; function computeNewHighWaterMark(n) { if (n >= MAX_HWM) { // حد 1 جيجابايت n = MAX_HWM } else { // قم بإزالة أعلى قوة تالية تبلغ 2 to منع زيادة hwm n--; n |= n >>> 1; n |= n >>> 4; > 16 ; ن++ } العودة ن؛}

نمط التدفق

تبدأ جميع التدفقات القابلة للقراءة في وضع الإيقاف المؤقت ويمكن تحويلها إلى وضع التدفق من خلال الطرق التالية:

إضافة معالج حدث "بيانات"؛ استدعاء طريقة "الاستئناف"؛ استخدم طريقة "الأنبوب" لإرسال البيانات إلى الدفق القابل للكتابة

في وضع التدفق، سيتم إخراج البيانات الموجودة في تجمع المخزن المؤقت تلقائيًا إلى المستهلك للاستهلاك، وفي الوقت نفسه، بعد كل إخراج بيانات، سيتم استدعاء طريقة _read تلقائيًا مرة أخرى لوضع البيانات من مصدر البيانات في تجمع المخزن المؤقت. إذا لم يكن هناك بيانات في تجمع المخزن المؤقت، فسيتم تمرير البيانات مباشرة إلى حدث البيانات دون المرور عبر تجمع ذاكرة التخزين المؤقت؛ حتى يتحول وضع التدفق إلى أوضاع الإيقاف المؤقت الأخرى، أو تتم قراءة البيانات من مصدر البيانات (اضغط (باطل))؛

يمكن تحويل التدفقات القابلة للقراءة مرة أخرى إلى وضع الإيقاف المؤقت عبر:

إذا لم يكن هناك هدف لخط الأنابيب، فسيتم استدعاء الدالةstream.pause()‎ . إذا كانت هناك أهداف لخطوط الأنابيب، فسيتم إزالة كافة أهداف خطوط الأنابيب. يمكن إزالة أهداف أنابيب متعددة عن طريق استدعاء الدفق الدفق.unpipe(). const { Readable } = require('stream')let count = 1000const myReadable = new Readable({ HighWaterMark: 300, read(size) { Letunk = null setTimeout(() => { if (count > 0) { LetchunkLength = Math.min(count, size)chunk = '1'.repeat(chunkLength) count -=chunkLength } this.push(chunk) }, 500) }})myReadable.on('data', data => { console .log(data.toString())})

قابل للكتابة

بالمقارنة مع التدفقات القابلة للقراءة، فإن التدفقات القابلة للكتابة أبسط.

عندما يستدعي المنتج الكتابة (القطعة)، فإنه سيختار داخليًا ما إذا كان سيتم تخزينها مؤقتًا في قائمة انتظار المخزن المؤقت أو استدعاء _write بناءً على بعض الحالات (الفلين، والكتابة، وما إلى ذلك)، بعد كل مرة تتم فيها كتابة البيانات، سيحاول مسحها البيانات في قائمة انتظار ذاكرة التخزين المؤقت. إذا تجاوز حجم البيانات في قائمة انتظار المخزن المؤقت القيمة العائمة (highWaterMark)، فسوف يُرجع المستهلك خطأ بعد استدعاء الكتابة (chunk). في هذا الوقت، يجب على المنتج التوقف عن الكتابة.

إذن متى يمكنني الاستمرار في الكتابة؟ عندما تتم كتابة جميع البيانات الموجودة في المخزن المؤقت بنجاح، سيتم تشغيل حدث التصريف بعد مسح قائمة انتظار المخزن المؤقت. في هذا الوقت، يمكن للمنتج الاستمرار في كتابة البيانات.

عندما يحتاج المنتج إلى الانتهاء من كتابة البيانات، فإنه يحتاج إلى استدعاء التابعstream.end لإعلامه بنهاية الدفق القابل للكتابة.

const { Writable, Duplex } = require('stream')let fileContent = ''const myWritable = new Writable({ HighWaterMark: 10, write(chunk, encoding, callback) {// سيتم استخدامه كأسلوب _write setTimeout(()) = >{ fileContent += Chunk callback()// يتم الاتصال به بعد اكتمال الكتابة}, 500) }})myWritable.on('Close', ()=>{ console.log('Close', fileContent)})myWritable write('123123')// truemyWritable.write('123123')// falsemyWritable.end() .

لاحظ أنه بعد أن تصل البيانات الموجودة في تجمع ذاكرة التخزين المؤقت إلى القيمة العائمة، قد تكون هناك عقد متعددة في تجمع ذاكرة التخزين المؤقت في هذا الوقت أثناء عملية مسح تجمع ذاكرة التخزين المؤقت (استدعاء دوري _قراءة)، لن تستهلك نفس الطول يتم استهلاك بيانات القيمة العائمة من خلال عقدة عازلة واحدة في كل مرة، حتى لو كان طول المخزن المؤقت غير متوافق مع القيمة العائمة.

const { Writable } = require('stream')let fileContent = ''const myWritable = new Writable({ HighWaterMark: 10, write(chunk, encoding, callback) { setTimeout(()=>{ fileContent += Chunk console.log ('الاستهلاك',chunk.toString()) رد الاتصال()// يتم الاتصال به بعد اكتمال الكتابة}, 100) }})myWritable.on('إغلاق', ()=>{ console.log('إغلاق', fileContent )})let count = 0function productionData(){ Let flag = true while (count <= 20 && flag){ flag = myWritable.write(count.toString()) count++ } if(count > 20){ myWritable.end( ) }}productionData()myWritable.on('drain', productionData)

ما سبق هو دفق قابل للكتابة بقيمة عائمة 10. الآن مصدر البيانات عبارة عن سلسلة أرقام مستمرة من 0 إلى 20، ويتم استخدام بيانات الإنتاج لكتابة البيانات.

أولاً، عند استدعاء myWritable.write("0") لأول مرة، نظرًا لعدم وجود بيانات في تجمع ذاكرة التخزين المؤقت، لا يدخل "0" إلى تجمع ذاكرة التخزين المؤقت، ولكن يتم إعطاؤه مباشرة إلى _wirte القيمة المرجعة لـ myWritable .write("0") صحيح

عند تنفيذ myWritable.write("1")، لأنه لم يتم استدعاء رد الاتصال _wirte بعد، فهذا يشير إلى أن البيانات الأخيرة لم تتم كتابتها بعد، ويضمن ترتيب كتابة البيانات فقط لتخزين "1" أضف إلى تجمع ذاكرة التخزين المؤقت. وهذا صحيح بالنسبة لل2-9 القادمة

عند تنفيذ myWritable.write("10")، يكون طول المخزن المؤقت 9 (1-9) ولم يصل بعد إلى القيمة العائمة، وتستمر إضافة "10" إلى تجمع ذاكرة التخزين المؤقت كمخزن مؤقت، وتجمع ذاكرة التخزين المؤقت يصبح الطول 11، لذلك يقوم myWritable.write("1") بإرجاع خطأ، مما يعني أن البيانات الموجودة في المخزن المؤقت كافية، ونحن بحاجة إلى انتظار إشعار حدث الاستنزاف لإنتاج البيانات مرة أخرى.

بعد 100 مللي ثانية، يتم استدعاء رد الاتصال _write("0"، التشفير، رد الاتصال) للإشارة إلى كتابة "0". ثم سيتحقق مما إذا كانت هناك بيانات في تجمع ذاكرة التخزين المؤقت، وإذا كانت موجودة، فسيقوم أولاً باستدعاء _read لاستهلاك العقدة الرئيسية لتجمع ذاكرة التخزين المؤقت ("1")، ثم يستمر في تكرار هذه العملية حتى يصبح تجمع ذاكرة التخزين المؤقت فارغًا. ، قم بتشغيل حدث التصريف، ثم قم بتنفيذ بيانات الإنتاج مرة أخرى.

اتصل بـ myWritable.write("11") لبدء العملية بدءًا من الخطوة 1 حتى نهاية الدفق.

دوبلكس

بعد فهم الدفق القابل للقراءة والدفق القابل للكتابة، من السهل فهم الدفق المزدوج، يرث الدفق المزدوج فعليًا الدفق القابل للقراءة ثم ينفذ الدفق القابل للكتابة (تتم كتابة الكود المصدري بهذه الطريقة، ولكن يجب أن يقال أنه تم تنفيذه). في نفس الوقت من الأفضل أن يكون لديك تدفقات قابلة للقراءة والكتابة).

يحتاج التدفق المزدوج إلى تنفيذ الطريقتين التاليتين في نفس الوقت

قم بتنفيذ طريقة _read() لإنتاج بيانات للتدفقات القابلة للقراءة

قم بتنفيذ طريقة _write() لاستهلاك البيانات للتدفقات القابلة للكتابة

لقد تم تقديم كيفية تنفيذ الطريقتين المذكورتين أعلاه في التدفقات القابلة للكتابة والقابلة للقراءة أعلاه. ما يجب ملاحظته هنا هو أن هناك مجموعتين مستقلتين من مجموعات المخزن المؤقت للتدفقات المزدوجة على التوالي، كما أن مصادر البيانات الخاصة بهما ليست هي نفسها.

خذ دفق الإدخال والإخراج القياسي لـ NodeJs كمثال:

عندما نقوم بإدخال البيانات في وحدة التحكم، يتم تشغيل حدث البيانات الخاص بها، مما يثبت أن لديها وظيفة دفق قابل للقراءة في كل مرة يقوم فيها المستخدم بإدخال، فإن ذلك يعادل استدعاء طريقة الدفع القابلة للقراءة لدفع البيانات المنتجة. عندما نستدعي طريقة الكتابة الخاصة بها، يمكننا أيضًا إخراج المحتوى إلى وحدة التحكم، ولكن لن يتم تشغيل حدث البيانات. وهذا يدل على أنه يحتوي على وظيفة دفق قابل للكتابة ويحتوي على مخزن مؤقت مستقل السماح لوحدة التحكم بعرض النص. // عندما يقوم المستخدم بإدخال البيانات على وحدة التحكم (_read)، سيتم تشغيل حدث البيانات، وهي إحدى سمات الدفق القابل للقراءةprocess.stdin.on('data', data=>{process.stdin.write(data })// إنتاج البيانات إلى دفق الإدخال القياسي كل ثانية (هذه إحدى ميزات الدفق القابل للكتابة، والتي سيتم إخراجها مباشرة إلى وحدة التحكم) ولن تؤدي إلى تشغيل datasetInterval(()=>{process.stdin.write) ("ليست بيانات تم إدخالها بواسطة وحدة تحكم المستخدم")}، 1000)

تحويل

يمكن اعتبار التدفق المزدوج بمثابة تدفق قابل للقراءة مع تدفق قابل للكتابة. كلاهما مستقلان، ولكل منهما مخازن داخلية مستقلة. أحداث القراءة والكتابة تحدث بشكل مستقل.

الدفق المزدوج ------------------| اقرأ <----- المصدر الخارجي أنت ------------------| اكتب -----> حوض خارجي ------------------|

تدفقات التحويل هي تدفقات مزدوجة، حيث تحدث عمليات القراءة والكتابة في علاقة السبب والنتيجة. ترتبط نقاط النهاية للتيار المزدوج من خلال بعض التحويلات. القراءة تتطلب حدوث الكتابة.

تحويل الدفق --------------|-------------- أنت تكتب ----> ----> أقرأك ----- ----------|--------------|

لإنشاء تدفقات التحويل، الشيء الأكثر أهمية هو تنفيذ طريقة _transform بدلاً من _write أو _read. في _transform، تتم معالجة (استهلاك) البيانات المكتوبة بواسطة الدفق القابل للكتابة ثم يتم إنتاج البيانات للدفق القابل للقراءة.

غالبًا ما تنفذ تدفقات التحويل طريقة ``_flush``، والتي سيتم استدعاؤها قبل نهاية الدفق، وتُستخدم بشكل عام لإلحاق شيء ما بنهاية الدفق. على سبيل المثال، تتم إضافة بعض معلومات الضغط هنا عند ضغط الملفات } = require('fs')const { Transform, PassThrough } = require('stream')const reurce = '1312123213124341234213423428354816273513461891468186499126412'const تحويل = تحويل جديد({ HighWaterMark: 10, تحويل(chunk,encoding, call رجوع) {// تحويل البيانات، استدعاء الدفع لإضافة نتيجة التحويل إلى تجمع ذاكرة التخزين المؤقت this.push(chunk.toString().replace('1', '@')) callback() }, Flush(callback){// تنفيذ this.push (' قبل مشغلات النهاية <<<') رد الاتصال () }})// الكتابة بشكل مستمر يكتب البيانات Let count = 0transform.write('>>>')function productionData() { Let flag = true while (count <= 20 && flag) { flag = Transform.write(count.toString()) count++ } if (count > 20) { converter.end() }}productionData()transform.on('drain', productionData)let result = '' Transform.on( 'data', data=>{ result += data.toString()})transform.on('end', ()=>{ console.log(result) // >>>0@23456789@ 0@1@ 2@3@4@5@6@7@8@920<<<})