Streams API

The Streams API in BanglaCode enables efficient processing of large amounts of data by breaking it into smaller chunks. Instead of loading entire files or datasets into memory at once, streams allow you to process data piece by piece, making your applications more memory-efficient and responsive.

Quick Start

// Create a writable stream
dhoro stream = stream_writable_srishti();

// Write data to stream
stream_lekho(stream, "Hello ");
stream_lekho(stream, "World!");

// Close stream
stream_bondho(stream);

dekho("Data:", buffer_text(buffer_theke(stream.Buffer)));

Core Concepts

Readable Streams (স্ট্রীম পড়ার)

Readable streams represent a source of data that you can read from chunk by chunk. They're ideal for processing large files or data sources without loading everything into memory.

Writable Streams (স্ট্রীম লেখার)

Writable streams represent a destination where you can write data chunk by chunk. They're perfect for creating files, sending data over networks, or any operation that produces data incrementally.

Event-Driven Processing

Streams emit events (data, end, error) that you can listen to, enabling reactive data processing patterns. This makes streams perfect for real-time data processing and transformations.

Backpressure Management

Streams automatically handle backpressure using the high water mark, preventing memory overflow when the producer is faster than the consumer. Default high water mark is 16KB.

API Reference

stream_readable_srishti(highWaterMark?)

Creates a new readable stream.

Parameters:

highWaterMark (optional): Maximum buffer size in bytes (default: 16384)

Returns:

Stream object with type "readable"

dhoro stream = stream_readable_srishti();
dhoro largeStream = stream_readable_srishti(65536); // 64KB buffer

stream_writable_srishti(highWaterMark?)

Creates a new writable stream.

Parameters:

highWaterMark (optional): Maximum buffer size in bytes (default: 16384)

Returns:

Stream object with type "writable"

dhoro stream = stream_writable_srishti();
dhoro customStream = stream_writable_srishti(32768); // 32KB buffer

stream_lekho(stream, data)

Writes data to a writable stream. Triggers "data" event handlers if registered.

Parameters:

stream: Writable stream object
data: String or Buffer to write

Returns:

True if buffer is below high water mark, false otherwise

dhoro stream = stream_writable_srishti();
stream_lekho(stream, "Hello World");
stream_lekho(stream, buffer_theke([72, 105])); // Write buffer

stream_poro(stream, size?)

Reads data from a readable stream.

Parameters:

stream: Readable stream object
size (optional): Number of bytes to read (default: all available)

Returns:

Buffer containing the read data, or null if stream is ended

dhoro stream = stream_readable_srishti();
dhoro data = stream_poro(stream);      // Read all
dhoro chunk = stream_poro(stream, 100); // Read 100 bytes

stream_bondho(stream)

Closes a stream, preventing further writes or reads. Triggers "end" event handlers.

Parameters:

stream: Stream object to close

dhoro stream = stream_writable_srishti();
stream_lekho(stream, "Final data");
stream_bondho(stream); // Close stream

stream_shesh(stream)

Signals that a readable stream has ended (no more data will be written to it). Triggers "end" event handlers.

Parameters:

stream: Readable stream to end

dhoro stream = stream_readable_srishti();
// ... produce data ...
stream_shesh(stream); // Signal end of data

stream_pipe(readable, writable)

Pipes data from a readable stream to a writable stream, automatically handling backpressure.

Parameters:

readable: Source readable stream
writable: Destination writable stream

dhoro source = stream_readable_srishti();
dhoro destination = stream_writable_srishti();
stream_pipe(source, destination);

stream_on(stream, eventName, handler)

Registers an event handler for stream events (data, end, error).

Parameters:

stream: Stream object
eventName: Event name ("data", "end", or "error")
handler: Function to call when event occurs

dhoro stream = stream_writable_srishti();

stream_on(stream, "data", kaj(chunk) {
  dekho("Received:", chunk);
});

stream_on(stream, "end", kaj() {
  dekho("Stream ended");
});

stream_on(stream, "error", kaj(err) {
  dekho("Error:", err);
});

Real-World Examples

Example 1: Large File Processing

Process a large file in chunks instead of loading it all into memory:

// Process large log file line by line
dhoro logStream = stream_writable_srishti();
dhoro errorCount = 0;
dhoro warningCount = 0;

// Register data handler to process chunks
stream_on(logStream, "data", kaj(chunk) {
  dhoro lines = bibhajan(chunk, "\n");
  
  ghuriye (dhoro i = 0; i < dorghyo(lines); i = i + 1) {
    dhoro line = lines[i];
    
    jodi (khuje(line, "ERROR") != mittha) {
      errorCount = errorCount + 1;
    } nahole jodi (khuje(line, "WARNING") != mittha) {
      warningCount = warningCount + 1;
    }
  }
});

// Handle end of stream
stream_on(logStream, "end", kaj() {
  dekho("Processing complete!");
  dekho("Errors:", errorCount);
  dekho("Warnings:", warningCount);
});

// Read file and write to stream in chunks
dhoro content = poro("large_log.txt");
dhoro chunkSize = 8192; // 8KB chunks

ghuriye (dhoro i = 0; i < dorghyo(content); i = i + chunkSize) {
  dhoro end = i + chunkSize;
  jodi (end > dorghyo(content)) {
    end = dorghyo(content);
  }
  dhoro chunk = angsho(content, i, end);
  stream_lekho(logStream, chunk);
}

stream_bondho(logStream);

Example 2: Data Transformation Pipeline

Create a pipeline to transform and filter data in real-time:

// Transform and filter data pipeline
dhoro inputStream = stream_writable_srishti();
dhoro outputStream = stream_writable_srishti();
dhoro transformedCount = 0;

// Transform: uppercase + filter (length > 5)
stream_on(inputStream, "data", kaj(chunk) {
  dhoro words = bibhajan(chunk, " ");
  dhoro transformed = [];
  
  ghuriye (dhoro i = 0; i < dorghyo(words); i = i + 1) {
    dhoro word = words[i];
    
    // Filter: only process words longer than 5 chars
    jodi (dorghyo(word) > 5) {
      dhoro upper = boro_hater(word);
      transformed = dhaaka(transformed, upper);
      transformedCount = transformedCount + 1;
    }
  }
  
  // Write transformed data to output
  jodi (dorghyo(transformed) > 0) {
    dhoro result = joro(transformed, " ");
    stream_lekho(outputStream, result + " ");
  }
});

// Handle output stream data
stream_on(outputStream, "data", kaj(chunk) {
  dekho("Transformed:", chunk);
});

// Handle completion
stream_on(inputStream, "end", kaj() {
  stream_bondho(outputStream);
  dekho("Transformation complete!");
  dekho("Processed words:", transformedCount);
});

// Process input data
dhoro input = "hello world banglacode programming language";
stream_lekho(inputStream, input);
stream_bondho(inputStream);

Example 3: Network Data Streaming

Stream data from network to file efficiently:

// Download and process data in streaming fashion
proyash kaj downloadAndProcess() {
  dhoro outputStream = stream_writable_srishti();
  dhoro bytesProcessed = 0;
  
  // Monitor data as it arrives
  stream_on(outputStream, "data", kaj(chunk) {
    bytesProcessed = bytesProcessed + dorghyo(chunk);
    
    // Show progress every 10KB
    jodi (bytesProcessed % 10240 == 0) {
      dekho("Downloaded:", bytesProcessed, "bytes");
    }
  });
  
  // Handle completion
  stream_on(outputStream, "end", kaj() {
    dekho("Download complete!");
    dekho("Total bytes:", bytesProcessed);
  });
  
  // Simulate network data (in real app, use HTTP client)
  dhoro data1 = "First chunk of data...";
  dhoro data2 = "Second chunk of data...";
  dhoro data3 = "Third chunk of data...";
  
  // Write chunks as they arrive
  stream_lekho(outputStream, data1);
  opekha ghumaao(100);
  
  stream_lekho(outputStream, data2);
  opekha ghumaao(100);
  
  stream_lekho(outputStream, data3);
  stream_bondho(outputStream);
  
  // Save to file
  lekho("downloaded.txt", buffer_text(buffer_theke(outputStream.Buffer)));
}

downloadAndProcess();

Best Practices

✅ DO: Always Close Streams

Always close streams when you're done with them to free resources. Use stream_bondho() or stream_shesh() to properly terminate streams.

✅ DO: Handle Events

Register handlers for "data", "end", and "error" events to create robust stream processing. This makes your code reactive and easier to maintain.

✅ DO: Use Appropriate Buffer Sizes

Set high water marks based on your data size. Default is 16KB, but adjust for large files (64KB-1MB) or small packets (4KB-8KB) for optimal performance.

✅ DO: Process Data in Chunks

Break large operations into smaller chunks. This prevents memory overflow and keeps your application responsive while processing large datasets.

❌ DON'T: Write to Closed Streams

Attempting to write to a closed stream will result in an error. Always check stream state or handle errors properly in production code.

❌ DON'T: Ignore Backpressure

Respect the return value of stream_lekho(). If it returns false, the buffer is full - pause writes until drained to avoid memory issues.

❌ DON'T: Mix Sync and Async

Don't mix synchronous file reads with asynchronous stream processing. Keep your data flow consistent - either all sync or all async.

❌ DON'T: Load Everything First

Avoid loading entire files into memory before streaming. That defeats the purpose! Use streams from the start for true streaming performance.

Performance Tips

🚀

Optimal Chunk Sizes

For file I/O: 8KB-64KB chunks work best. For network data: 4KB-16KB. For large datasets: 64KB-1MB. Test your specific use case to find the sweet spot.

💾

Memory Efficiency

Streams keep memory usage constant regardless of data size. A 10GB file uses only ~16KB RAM when processed with default streams - that's a 625,000x improvement!

⚡

Parallel Processing

Combine streams with Worker Threads for parallel data processing. Each worker can process its own stream for maximum throughput.

🔄

Pipeline Efficiency

Use stream_pipe() to create efficient data pipelines. The piping mechanism handles backpressure automatically, optimizing throughput.

Common Patterns

Pattern 1: Transform Stream

// Create reusable transform stream
kaj transformStream(inputStream, transformFn) {
  dhoro outputStream = stream_writable_srishti();
  
  stream_on(inputStream, "data", kaj(chunk) {
    dhoro transformed = transformFn(chunk);
    stream_lekho(outputStream, transformed);
  });
  
  stream_on(inputStream, "end", kaj() {
    stream_bondho(outputStream);
  });
  
  ferao outputStream;
}

// Usage
dhoro input = stream_writable_srishti();
dhoro output = transformStream(input, kaj(data) {
  ferao boro_hater(data); // Uppercase transform
});

Pattern 2: Buffered Reader

// Read file in fixed-size chunks
kaj readInChunks(filepath, chunkSize, callback) {
  dhoro content = poro(filepath);
  dhoro stream = stream_writable_srishti();
  
  stream_on(stream, "data", callback);
  
  ghuriye (dhoro i = 0; i < dorghyo(content); i = i + chunkSize) {
    dhoro end = i + chunkSize;
    jodi (end > dorghyo(content)) {
      end = dorghyo(content);
    }
    stream_lekho(stream, angsho(content, i, end));
  }
  
  stream_bondho(stream);
}

// Usage
readInChunks("data.txt", 1024, kaj(chunk) {
  dekho("Processing chunk:", dorghyo(chunk), "bytes");
});

Pattern 3: Stream Aggregator

// Aggregate stream data
kaj aggregateStream(stream, reduceFn, initialValue) {
  dhoro accumulated = initialValue;
  
  stream_on(stream, "data", kaj(chunk) {
    accumulated = reduceFn(accumulated, chunk);
  });
  
  stream_on(stream, "end", kaj() {
    dekho("Final result:", accumulated);
  });
}

// Usage: Count total characters
dhoro stream = stream_writable_srishti();
aggregateStream(stream, kaj(total, chunk) {
  ferao total + dorghyo(chunk);
}, 0);

stream_lekho(stream, "Hello ");
stream_lekho(stream, "World");
stream_bondho(stream);

Streams vs Regular I/O

❌ Without Streams (Bad for Large Files)

// Loads entire 10GB file into memory!
dhoro content = poro("huge.log");

dhoro lines = bibhajan(content, "\n");
dhoro errors = 0;

ghuriye (dhoro i = 0; i < dorghyo(lines); i = i + 1) {
  jodi (khuje(lines[i], "ERROR") != mittha) {
    errors = errors + 1;
  }
}

dekho("Errors:", errors);

// Memory usage: 10GB+
// Time: Very slow (disk → RAM all at once)

✅ With Streams (Optimal)

// Processes 10GB file with only 16KB RAM!
dhoro stream = stream_writable_srishti();
dhoro errors = 0;

stream_on(stream, "data", kaj(chunk) {
  dhoro lines = bibhajan(chunk, "\n");
  ghuriye (dhoro i = 0; i < dorghyo(lines); i = i + 1) {
    jodi (khuje(lines[i], "ERROR") != mittha) {
      errors = errors + 1;
    }
  }
});

// ... write chunks to stream ...

// Memory usage: ~16KB constant
// Time: Much faster (streaming)

Related APIs

Buffer API

Work with binary data efficiently. Streams use Buffers internally for data storage.

Worker Threads

Process streams in parallel across multiple CPU cores for maximum performance.

EventEmitter

Streams use events (data, end, error) for reactive processing patterns.

Summary

The Streams API in BanglaCode provides a powerful and memory-efficient way to process large amounts of data. By breaking data into smaller chunks and processing them incrementally, streams enable you to build scalable applications that handle files and datasets of any size.

Key benefits:

Memory Efficiency: Process gigabytes of data with only kilobytes of RAM
Performance: Start processing immediately without waiting for entire files to load
Backpressure: Automatic flow control prevents memory overflow
Event-Driven: React to data as it arrives for real-time processing
Composability: Chain streams together to create data processing pipelines

Use streams whenever you're working with large files, network data, or any scenario where you want to process data incrementally rather than all at once.