Music Streamer

music-streamer.js

const fs = require('fs');
const stream = require('stream');
const icy = require('icy');
const express = require('express');

class BroadcastStream {
  readable;
  sinks = [];
  
  subscribe() {
    const sink = stream.PassThrough({
      readableHighWaterMark: 48 * 1000,
      writableHighWaterMark: 48 * 1000,
    });
    this.sinks.push(sink);
    return sink;
  }

  attachReadable(readable) {
    this.readable = readable;
    this.readable.count = 0;

    this.readable.on('data', (chunk) => {
      this.readable.count += 1;
      const chunkId = this.readable.count;

      setTimeout(() => {
        for (let index in this.sinks) {
          console.log(`Send chunk #${chunkId} to sink #${index}`)
          this.sinks[index].write(chunk);
        }
      }, chunkId * 1000);
    });
  }
}

class Streamer extends stream.PassThrough {
  app;
  metaint = 48 * 1000;
  metadata;

  constructor(app) {
    super({
      highWaterMark: 48 * 1000,
    });

    this.broadcast = new BroadcastStream();
    this.broadcast.attachReadable(this);

    this.app = app;
    this.app.disable('x-powered-by');
    this.setupRouting();
  }

  setupRouting() {
    this.app.get('/listen', (req, res) => this.listener(req, res));
  }

  listener(req, res) {
    const headers = {
      "Content-Type": "audio/mpeg",
      "Connection": "close",
      "icy-metaint": this.metaint,
    };
    res.writeHead(200, headers);

    const icyWriter = icy.Writer(this.metaint);
    const sink = this.broadcast.subscribe();

    icyWriter.pipe(sink, { end: false });
    sink.pipe(res, { end: false });

    req.connection.on('close', () => {
      this.unpipe(icyWriter);
    });
  }
}

const readStream = fs.createReadStream('test.mp3', { highWaterMark: 48 * 1000 });
const app = express();
app.listen(3000);

const streamer = new Streamer(app);
readStream.pipe(streamer);