abelcastro.dev

Recently, I have been learning how to develop REST APIs using Express and how to write tests for them. As part of my learning journey, I have also been exploring TypeScript ORMs capable of handling streams.

A stream is a sequence of data elements made available over time, used for handling large volumes of data or continuous data flow. Streams are useful because they allow for efficient processing of data chunks as soon as they are available, without waiting for the entire data set. This is particularly helpful in managing resources and improving performance in applications like real-time data feeds, file processing, and network communications.

Streams can be particularly useful in various scenarios, such as:

• File Processing: When handling large files, streams can read and process data in chunks, preventing memory overload and enabling efficient data handling.
• Network Communications: Streams are ideal for handling data over network requests and responses, such as in web APIs, where data can be processed as it arrives rather than waiting for the complete set.
• Real-time Data Feeds: For applications like social media feeds, stock tickers, or live sports updates, streams allow for real-time processing and display of incoming data.
• Audio/Video Streaming: Media streaming services use streams to deliver content in a buffered manner, enabling smooth playback while the rest of the data is being received.
• Data Transformation: In ETL (Extract, Transform, Load) operations, streams can efficiently process and transform data before it's loaded into a database or another system.
• Log Processing: For applications that generate large volumes of logs, streams can help in efficiently reading and processing log data for monitoring or analytics.
• Chat Applications: Streams enable the implementation of real-time messaging features in chat applications, handling continuous data flow between users.

For this purpose, I created the blog-orm repository, which showcases how to interact with a PostgreSQL database using TypeORM and return data as a stream.

For creating the DB Post table you just need to implement a Post.ts entity and data-source.ts:

//entity/Post.ts
import { Entity, PrimaryGeneratedColumn, Column } from "typeorm"

@Entity()
export class Post {
    @PrimaryGeneratedColumn()
    id: number

    @Column()
    title: string

    @Column()
    content: string

    @Column()
    isPublished: boolean
}

//data-source.ts
import "reflect-metadata"
import { DataSource } from "typeorm"
import { Post } from "./entity/Post"

export const AppDataSource = new DataSource({
    type: "postgres",
    host: "localhost",
    port: 5432,
    username: "postgres",
    password: "postgres",
    database: "blog",
    synchronize: true,
    logging: false,
    entities: [Post],
    migrations: [],
    subscribers: [],
})

A service could look like that:

//postService.ts
import { AppDataSource } from "./data-source";
import { Post } from "./entity/Post";

export async function getAllStream() {
    return await AppDataSource.getRepository(Post).createQueryBuilder("post").stream();
}

And finally the controller:

//controller.ts
import { Request, Response, NextFunction } from "express";
import * as postService from "./postService"
import { pipeline } from "stream/promises";
import Disassembler from 'stream-json/Disassembler';
import Stringer from 'stream-json/Stringer';

export async function getAllStream(req: Request, res: Response, next: NextFunction) {
    try {
        const stream = await postService.getAllStream();
        res.setHeader('Content-Type', 'application/json');
        const tokenizer = new Disassembler();
        const jsonStream = new Stringer({ makeArray: true });
        pipeline(stream, tokenizer, jsonStream, res)
    } catch (error) {
        console.error(error);
        res.status(500).send('An error occurred');
    }
}

And that's the whole magic! You can find the complete code in this repository.