Introduction to Node.js Streams

When building tools that process code—like formatters, linters, or transpilers—efficiency and scalability are crucial. Node.js streams provide an elegant solution for handling large files and complex processing pipelines without overwhelming system resources.

In this guide, we'll explore how to build a unified code formatter using Node.js streams. This approach allows us to process code in chunks, transform it incrementally, and handle files of any size with consistent memory usage.

Why Use Streams for Formatting?

Traditional code formatters often load entire files into memory, which can be problematic when dealing with large codebases. Stream-based formatters offer several advantages:

• Memory efficiency: Process code in chunks rather than loading entire files
• Scalability: Handle files of any size with consistent performance
• Composability: Chain multiple transformations together in a pipeline
• Backpressure handling: Automatically manage processing speed across the pipeline
• Parallelization: Process multiple files concurrently with controlled resource usage

While stream-based processing adds some complexity, the benefits make it worthwhile for building robust, production-grade formatting tools.

Node.js Stream Basics

Before diving into formatter implementation, let's review the fundamentals of Node.js streams.

Types of Streams

Node.js provides four fundamental types of streams:

• Readable: Sources that you can read data from (e.g., file input)
• Writable: Destinations that you can write data to (e.g., file output)
• Duplex: Both readable and writable (e.g., network sockets)
• Transform: Duplex streams that modify data as it's written and read (perfect for formatters)

For our formatter, we'll primarily use Transform streams to process code as it flows through our pipeline.

Transform Streams

Transform streams are the workhorses of our formatter. Here's a basic example:

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

class SimpleTransformer extends Transform {
  constructor(options = {}) {
    super(options);
  }

  _transform(chunk, encoding, callback) {
    // Process the chunk of data
    const transformedChunk = someTransformation(chunk);
    
    // Push the transformed chunk to the output
    this.push(transformedChunk);
    
    // Signal that we're done processing this chunk
    callback();
  }
}

// Usage
const transformer = new SimpleTransformer();
sourceStream.pipe(transformer).pipe(destinationStream);

The _transform method is where the magic happens. It receives chunks of data, processes them, and pushes the transformed chunks to the output stream.

The Stream Pipeline

For robust error handling, we'll use the pipeline{" "} function instead of chaining pipe calls:

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

pipeline(
  fs.createReadStream('input.js'),
  new LexerStream(),
  new ParserStream(),
  new FormatterStream(),
  fs.createWriteStream('output.js'),
  (err) => {
    if (err) {
      console.error('Pipeline failed:', err);
    } else {
      console.log('Formatting complete');
    }
  }
);

This approach properly propagates errors and ensures resources are cleaned up, even if an error occurs midway through the pipeline.

Formatter Architecture

Our unified formatter will follow a modular architecture with several key components.

Core Components

The formatter consists of these primary components:

1. Lexer Stream: Converts raw code text into tokens
2. Parser Stream: Transforms tokens into an Abstract Syntax Tree (AST)
3. Formatter Stream: Applies formatting rules to the AST
4. Output Stream: Generates formatted code from the modified AST

This separation of concerns makes the formatter extensible and maintainable.

Plugin System

To support multiple languages, we'll implement a plugin system:

class FormatterRegistry {
  constructor() {
    this.formatters = new Map();
  }

  register(language, formatter) {
    this.formatters.set(language, formatter);
  }

  getFormatter(language) {
    if (!this.formatters.has(language)) {
      throw new Error(`No formatter registered for language: \${language}`);
    }
    return this.formatters.get(language);
  }
}

// Usage
const registry = new FormatterRegistry();
registry.register('javascript', new JavaScriptFormatter());
registry.register('typescript', new TypeScriptFormatter());

const formatter = registry.getFormatter('javascript');

This approach allows users to add support for new languages without modifying the core formatter code.

Configuration Management

Formatters need flexible configuration options. We'll implement a configuration system that:

• Loads defaults from a base configuration
• Merges user-provided options
• Supports configuration files (.prettierrc, .eslintrc, etc.)
• Validates configuration values

class FormatterConfig {
  constructor(options = {}) {
    this.options = {
      ...this.getDefaultOptions(),
      ...options
    };
    this.validate();
  }

  getDefaultOptions() {
    return {
      indentSize: 2,
      indentStyle: 'space',
      lineWidth: 80,
      singleQuote: false,
      trailingComma: 'es5',
      // More default options...
    };
  }

  validate() {
    // Validate that all options have appropriate values
    if (typeof this.options.indentSize !== 'number' || this.options.indentSize < 0) {
      throw new Error('indentSize must be a non-negative number');
    }
    // More validation...
  }
}

// Usage
const config = new FormatterConfig({ indentSize: 4, singleQuote: true });

Implementation

Now, let's implement each component of our stream-based formatter.

Lexer Stream

The lexer stream converts raw code text into a stream of tokens:

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

class LexerStream extends Transform {
  constructor(language, options = {}) {
    super({ ...options, objectMode: true });
    this.language = language;
    this.buffer = '';
    this.lexer = this.getLexerForLanguage(language);
  }

  getLexerForLanguage(language) {
    // Return appropriate lexer based on language
    switch (language) {
      case 'javascript':
        return require('./lexers/javascript');
      case 'typescript':
        return require('./lexers/typescript');
      // Other languages...
      default:
        throw new Error(`Unsupported language: \${language}`);
    }
  }

  _transform(chunk, encoding, callback) {
    // Append the new chunk to our buffer
    this.buffer += chunk.toString();
    
    // Try to extract complete tokens
    let tokens;
    try {
      // The lexer might not consume the entire buffer if it ends mid-token
      const result = this.lexer.tokenize(this.buffer);
      tokens = result.tokens;
      this.buffer = result.remainder;
      
      // Push each token as a separate object
      tokens.forEach(token => this.push(token));
      
      callback();
    } catch (err) {
      callback(err);
    }
  }

  _flush(callback) {
    // Process any remaining buffer content
    if (this.buffer.trim().length > 0) {
      try {
        const result = this.lexer.tokenize(this.buffer, true);
        result.tokens.forEach(token => this.push(token));
        callback();
      } catch (err) {
        callback(err);
      }
    } else {
      callback();
    }
  }
}

Note that we're using objectMode: true because we're working with token objects rather than raw buffers.

Parser Stream

The parser stream converts tokens into an Abstract Syntax Tree (AST):

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

class ParserStream extends Transform {
  constructor(language, options = {}) {
    super({ ...options, objectMode: true });
    this.language = language;
    this.tokens = [];
    this.parser = this.getParserForLanguage(language);
  }

  getParserForLanguage(language) {
    // Return appropriate parser based on language
    switch (language) {
      case 'javascript':
        return require('./parsers/javascript');
      case 'typescript':
        return require('./parsers/typescript');
      // Other languages...
      default:
        throw new Error(`Unsupported language: \${language}`);
    }
  }

  _transform(token, encoding, callback) {
    // Collect tokens until we have enough to parse
    this.tokens.push(token);
    
    // Check if we have a complete parsable unit
    if (this.parser.canParse(this.tokens)) {
      try {
        const ast = this.parser.parse(this.tokens);
        this.push(ast);
        this.tokens = [];
        callback();
      } catch (err) {
        callback(err);
      }
    } else {
      // Need more tokens
      callback();
    }
  }

  _flush(callback) {
    // Parse any remaining tokens
    if (this.tokens.length > 0) {
      try {
        const ast = this.parser.parse(this.tokens, true);
        this.push(ast);
        callback();
      } catch (err) {
        callback(err);
      }
    } else {
      callback();
    }
  }
}

The parser accumulates tokens until it has enough to parse a complete syntax unit (like a statement or expression).

Formatter Stream

The formatter stream applies formatting rules to the AST:

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

class FormatterStream extends Transform {
  constructor(config, options = {}) {
    super({ ...options, objectMode: true });
    this.config = config;
    this.rules = this.loadRules(config);
  }

  loadRules(config) {
    // Load and configure formatting rules based on config
    return [
      new IndentationRule(config.indentSize, config.indentStyle),
      new LineWidthRule(config.lineWidth),
      new QuoteStyleRule(config.singleQuote),
      // More rules...
    ];
  }

  _transform(ast, encoding, callback) {
    try {
      // Apply each formatting rule to the AST
      let formattedAst = ast;
      for (const rule of this.rules) {
        formattedAst = rule.apply(formattedAst);
      }
      
      this.push(formattedAst);
      callback();
    } catch (err) {
      callback(err);
    }
  }
}

Each formatting rule is implemented as a separate class, making it easy to add, remove, or customize rules.

Output Stream

The output stream generates formatted code from the modified AST:

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

class OutputStream extends Transform {
  constructor(language, options = {}) {
    super({ ...options, objectMode: true });
    this.language = language;
    this.generator = this.getGeneratorForLanguage(language);
  }

  getGeneratorForLanguage(language) {
    // Return appropriate code generator based on language
    switch (language) {
      case 'javascript':
        return require('./generators/javascript');
      case 'typescript':
        return require('./generators/typescript');
      // Other languages...
      default:
        throw new Error(`Unsupported language: \${language}`);
    }
  }

  _transform(ast, encoding, callback) {
    try {
      // Generate code from the AST
      const code = this.generator.generate(ast);
      
      // Output as string
      this.push(code);
      callback();
    } catch (err) {
      callback(err);
    }
  }
}

The code generator converts the AST back into formatted code text, which can then be written to a file or returned to the user.

Adding Language Support

One of the key benefits of our unified formatter is the ability to support multiple languages through plugins.

JavaScript Support

Let's implement JavaScript support using a popular parser like Acorn:

// lexers/javascript.js
const acorn = require('acorn');

module.exports = {
  tokenize(code, isEnd = false) {
    // For simplicity, we're using acorn's full parser
    // In a real implementation, you might use a dedicated tokenizer
    try {
      const tokens = [];
      acorn.parse(code, {
        ecmaVersion: 'latest',
        onToken: token => tokens.push(token)
      });
      
      return {
        tokens,
        remainder: '' // Acorn processes the entire input
      };
    } catch (err) {
      if (isEnd) {
        throw err; // This is the end of the stream, so the error is real
      }
      
      // If not the end, we might have incomplete code
      // Return any tokens we could extract and keep the rest in the buffer
      return {
        tokens: [],
        remainder: code
      };
    }
  }
};

// parsers/javascript.js
const acorn = require('acorn');

module.exports = {
  canParse(tokens) {
    // Simplified: in reality, you'd check for complete statements
    return tokens.length > 0;
  },
  
  parse(tokens, isEnd = false) {
    // Reconstruct code from tokens
    const code = tokens.map(t => t.value).join('');
    
    try {
      // Parse the code into an AST
      return acorn.parse(code, {
        ecmaVersion: 'latest',
        locations: true
      });
    } catch (err) {
      if (isEnd) {
        throw err;
      }
      
      // If not the end, we might need more tokens
      return null;
    }
  }
};

// generators/javascript.js
const astring = require('astring');

module.exports = {
  generate(ast) {
    // Generate code from AST
    return astring.generate(ast, {
      indent: '  '
    });
  }
};

This implementation uses Acorn for parsing and Astring for code generation. In a real-world formatter, you might use more specialized tools or implement custom logic.

TypeScript Support

Adding TypeScript support is similar, but uses TypeScript-specific tools:

// lexers/typescript.js
const ts = require('typescript');

module.exports = {
  tokenize(code, isEnd = false) {
    const scanner = ts.createScanner(
      ts.ScriptTarget.Latest,
      /* skipTrivia */ false,
      ts.LanguageVariant.Standard,
      code
    );
    
    const tokens = [];
    let token = scanner.scan();
    
    while (token !== ts.SyntaxKind.EndOfFileToken) {
      tokens.push({
        type: token,
        value: scanner.getTokenText()
      });
      token = scanner.scan();
    }
    
    return {
      tokens,
      remainder: ''
    };
  }
};

// parsers/typescript.js
const ts = require('typescript');

module.exports = {
  canParse(tokens) {
    return tokens.length > 0;
  },
  
  parse(tokens, isEnd = false) {
    // Reconstruct code from tokens
    const code = tokens.map(t => t.value).join('');
    
    // Parse the TypeScript code
    const sourceFile = ts.createSourceFile(
      'temp.ts',
      code,
      ts.ScriptTarget.Latest,
      /* setParentNodes */ true
    );
    
    return sourceFile;
  }
};

// generators/typescript.js
const ts = require('typescript');

module.exports = {
  generate(ast) {
    const printer = ts.createPrinter({
      newLine: ts.NewLineKind.LineFeed
    });
    
    return printer.printFile(ast);
  }
};

This implementation uses the TypeScript compiler API for lexing, parsing, and code generation.

HTML and CSS Support

For HTML and CSS, we can use specialized parsers like parse5 and postcss:

const unified = require('unified')
const remarkParse = require('remark-parse')
const remarkStringify = require('remark-stringify')
const remarkGfm = require('remark-gfm')

const processor = unified()
  .use(remarkParse)
  .use(remarkGfm)
  .use(remarkStringify)

const markdown = '# Hello World\n\nThis is a test.'
const result = processor.processSync(markdown)
console.log(result.toString())

These examples are simplified. In a real implementation, you'd integrate these formatters into the stream pipeline architecture.

Custom Languages

For custom languages, you'll need to implement:

1. A lexer that converts code to tokens
2. A parser that converts tokens to an AST
3. Formatting rules specific to the language
4. A code generator that converts the AST back to formatted code

The plugin system makes it easy to add support for any language with appropriate tools.

Performance Considerations

Stream-based formatters offer excellent performance characteristics, but require careful implementation.

Memory Usage

To maintain low memory usage:

• Process code in reasonably sized chunks
• Avoid accumulating large data structures
• Release references to processed data
• Consider using WeakMap/WeakSet for caches

Monitor memory usage during development to identify and fix leaks:

const memoryUsage = () => {
  const used = process.memoryUsage();
  console.log(`Memory usage: \${Math.round(used.heapUsed / 1024 / 1024)} MB`);
};

// Call periodically to track memory usage
const interval = setInterval(memoryUsage, 1000);
// Clear when done
clearInterval(interval);

Handling Backpressure

Backpressure occurs when a slow consumer can't keep up with a fast producer. Node.js streams handle this automatically, but you should be aware of it:

// Check if the stream is ready for more data
if (!writableStream.write(chunk)) {
  // The buffer is full, wait for drain
  readableStream.pause();
  writableStream.once('drain', () => {
    // Resume reading when the buffer empties
    readableStream.resume();
  });
}

When using pipeline or pipe, this is handled for you, but custom stream implementations should respect backpressure signals.

Benchmarks

To evaluate formatter performance, benchmark key metrics: