Streams — الأنماط المتقدمة

const { Readable } = require('stream');

const readable = new Readable({
  read(size) {
    this.push(Math.random().toString());
    this.push(null); // انتهاء البيانات
  }
});

readable.on('data', chunk => {
  console.log('جملة:', chunk.toString());
});

const { Writable } = require('stream');

const writable = new Writable({
  write(chunk, encoding, callback) {
    console.log('كتابة:', chunk.toString());
    callback();
  }
});

writable.write('سلام');
writable.end();

const { Transform } = require('stream');

const upperCase = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

process.stdin.pipe(upperCase).pipe(process.stdout);

const { Duplex } = require('stream');

const duplex = new Duplex({
  read(size) {
    this.push('بيانات مقروءة');
    this.push(null);
  },
  write(chunk, encoding, callback) {
    console.log('بيانات مكتوبة:', chunk.toString());
    callback();
  }
});

const { Transform } = require('stream');

class JSONParser extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop(); // الباقي

    for (const line of lines) {
      try {
        this.push(JSON.parse(line));
      } catch (err) {
        this.destroy(err);
        return;
      }
    }
    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      try {
        this.push(JSON.parse(this.buffer));
      } catch (err) {
        this.destroy(err);
        return;
      }
    }
    callback();
  }
}

const parser = new JSONParser();
parser.on('data', obj => console.log(obj));
parser.write('{"name": "أحمد"}\n{"name": "سارة"}');
parser.end();

const { Readable, Writable } = require('stream');

const source = new Readable({
  highWaterMark: 16,
  read() {
    for (let i = 0; i < 100; i++) {
      const canContinue = this.push(Buffer.alloc(1024));
      if (!canContinue) {
        console.log('تم تطبيق backpressure عند chunk', i);
        break;
      }
    }
  }
});

const destination = new Writable({
  highWaterMark: 16,
  write(chunk, encoding, callback) {
    setTimeout(callback, 10); // محاكاة بطء
  }
});

source.pipe(destination);

const canContinue = writable.write(data);
if (!canContinue) {
  writable.once('drain', () => {
    console.log('المخزن فارغ، يمكن الاستمرار');
  });
}

const fs = require('fs');

const readStream = fs.createReadStream('input.txt');
const writeStream = fs.createWriteStream('output.txt');

readStream.pipe(writeStream);

readStream.on('error', err => console.error(err));
writeStream.on('error', err => console.error(err));

const { pipeline } = require('stream/promises');
const fs = require('fs');

async function safeCopy() {
  try {
    await pipeline(
      fs.createReadStream('input.txt'),
      fs.createWriteStream('output.txt')
    );
    console.log('تم النسخ بنجاح');
  } catch (err) {
    console.error('فشل النسخ:', err);
  }
}

const { Transform } = require('stream');

const filterStream = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    if (chunk.age > 18) {
      this.push(chunk);
    }
    callback();
  }
});

filterStream.on('data', user => console.log('مؤهل:', user.name));
filterStream.write({ name: 'خالد', age: 25 });
filterStream.write({ name: 'نور', age: 16 });
filterStream.end();

const fs = require('fs');
const readline = require('readline');

async function processFile(path) {
  const readStream = fs.createReadStream(path, { encoding: 'utf8' });
  const rl = readline.createInterface({ input: readStream });

  for await (const line of rl) {
    if (line.trim()) {
      console.log('سطر:', line);
    }
  }
}

processFile('data.txt').catch(console.error);

const { pipeline } = require('stream/promises');
const zlib = require('zlib');
const fs = require('fs');

async function compressFile(input, output) {
  try {
    await pipeline(
      fs.createReadStream(input),
      zlib.createGzip(),
      fs.createWriteStream(output)
    );
    console.log('تم الضغط بنجاح');
  } catch (err) {
    if (err.code === 'ENOENT') {
      console.error('الملف غير موجود:', input);
    } else if (err.code === 'Z_DATA_ERROR') {
      console.error('بيانات مضغوطة تالفة');
    } else {
      console.error('خطأ غير متوقع:', err);
    }
  }
}

const stream = fs.createReadStream('non-existent.txt');
stream.on('error', err => {
  if (err.code === 'ENOENT') {
    console.error('الملف غير موجود');
  } else {
    console.error('خطأ في القراءة:', err.message);
  }
});