client.ts

import socketIo = require('socket.io-client')

const client = socketIo.connect('http://localhost:8080')

setInterval(() => {
    client.emit('custom_stream', Buffer.from('aaaa'))
}, 1000)

socketIoStream.ts

import socketIo = require('socket.io')
import * as stream from 'stream'
import * as fs from 'fs'

const server = socketIo(8080)
server.on('connection', socket => {
    console.log('connected!')
    const readableStream = getReadableFromSocketIo(socket)
    readableStream.pipe(fs.createWriteStream('./testFile'))
})

function getReadableFromSocketIo(socketIoClientConnection: socketIo.Socket) {
    const readableStream = new stream.Readable()

    let buffer = Buffer.alloc(0)
    let finished = false
    let requestedSize = 0
    readableStream._read = size => {
        requestedSize += size
        writeToStream()
    }

    function writeToStream() {
        const amountToWrite = Math.min(requestedSize, buffer.length)
        if (amountToWrite > 0) {
            readableStream.push(buffer.slice(0, amountToWrite))
            requestedSize -= amountToWrite
            buffer = buffer.slice(amountToWrite)
        }

        if (finished && buffer.length === 0) {
            readableStream.push(null)
        }
    }

    socketIoClientConnection.on('custom_stream', chunk => {
        console.log('AAAA')
        const customCheckIfEnd: boolean = false
        if (customCheckIfEnd) {
            finished = true
            return
        }
        console.log(buffer.length)
        buffer = Buffer.concat([buffer, chunk])
        writeToStream()
    })

    return readableStream
}