nest-ndjson-req-stream

Accept and automatically parse NDJSON stream requests in NestJS with Express!

Description

A lightweight library that enables NestJS applications to accept and process streaming NDJSON (Newline Delimited JSON) requests with Express. Perfect for handling large datasets, real-time data feeds, and streaming APIs where each line contains a valid JSON object.

Features

• ✨ Simple decorator-based API for handling NDJSON streams
• 🚀 Memory-efficient processing using AsyncGenerators
• 🎯 TypeScript support with generic types
• 🔧 Configurable batch processing
• 🛡️ Automatic content-type validation
• ⚡ Zero dependencies (only NestJS peer dependencies)

Requirements

• Node.js >= 20.0.0
• NestJS >= 10.0.0
• Express framework

Installation

npm install nest-ndjson-req-stream

Quick Start

Simply use the decorator on request parameters in your controller

import { Controller, Post } from '@nestjs/common';
import { NdJsonStreamReq, NdJsonStreamRequest } from 'nest-ndjson-req-stream';

interface DataItem {
	id: number;
	name: string;
	value: number;
}

@Controller('stream')
export class StreamController {
	@Post('process')
	async processStream(
		@NdJsonStreamReq<DataItem>() request: NdJsonStreamRequest<DataItem>
	) {
		const results: DataItem[] = [];

		// Process each item from the stream
		for await (const item of request.body) {
			// Process your data here
			console.log('Received:', item);
			results.push(item);
		}

		return {
			message: 'Stream processed successfully',
			itemCount: results.length,
		};
	}
}

Error Handling

If the stream contains a malformed JSON line, request.body will throw at that point in the for await loop. Wrap your controller logic to return a 400:

@Post('process')
async processStream(
  @NdJsonStreamReq<DataItem>() request: NdJsonStreamRequest<DataItem>
) {
  try {
    for await (const item of request.body) {
      console.log('Received:', item);
    }
  } catch (error) {
    throw new BadRequestException(`Invalid NDJSON: ${error.message}`);
  }
}

The error object carries .line (the offending text), .itemNumber (1-based position in the stream), and .cause (the original SyntaxError).

Type Safety

Use TypeScript generics for type-safe stream processing:

interface User {
  id: string;
  email: string;
  profile: {
    name: string;
    age: number;
  };
}

@Post('users')
async importUsers(
  @NdJsonStreamReq() request: NdJsonStreamRequest<User>
) {
  for await (const user of request.body) {
    // TypeScript knows that user is of type User
    console.log(user.email); // ✅ Type safe
    console.log(user.profile.name); // ✅ Type safe
  }
}

API Reference

@NdJsonStreamReq(options?: NdJsonStreamOptions)

Parameter decorator for handling NDJSON streaming requests.

Options:

• batchSize?: number - The batch size for processing streamed objects (default: 25)

NdJsonStreamRequest<T>

Extended Express Request interface with AsyncGenerator type body:

• body: AsyncGenerator<T> - AsyncGenerator that yields parsed NDJSON objects
• batchSize: number - The configured batch size for processing

Testing

# Run unit tests
npm run test

# Run tests with coverage
npm run test:cov

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

• Fork the repository
• Create your feature branch (git checkout -b feature/AmazingFeature)
• Commit your changes (git commit -m 'Add some AmazingFeature')
• Push to the branch (git push origin feature/AmazingFeature)
• Open a Pull Request

License

This project is MIT licensed.