hypertube/hypertube.js

const fs = require('fs');
const b4a = require('b4a');
const Hyperswarm = require('hyperswarm');
const gracefulGoodbye = require('graceful-goodbye');
const crypto = require('hypercore-crypto');
let rand = Math.floor(Math.random() * 99999).toString();
let USERNAME = "annon" + rand

const stream = require('youtube-audio-stream')
const decoder = require('@suldashi/lame').Decoder
const Speaker = require('speaker');

let audioPlayer = new Speaker({
  channels: 2,
  bitDepth: 16,
  sampleRate: 44100,
});

// Flag to keep track of whether the stream is currently playing
let isPlaying = false;

function startStream(URL) {
  // Create a speaker instance to play the audio data
  // Stream the audio file
  const audioStream = stream(URL)
    .pipe(decoder());

  // Send the audio data to all connected peers
  audioStream.on('data', data => {
    for (const conn of conns) {
      conn.write(data);
    }
  });

  // Check if the audio player has ended
  if (!audioPlayer.writable) {
    // Create a new audio player if it has ended
    audioPlayer = new Speaker({
      channels: 2,
      bitDepth: 16,
      sampleRate: 44100,
    });
    audioStream.pipe(audioPlayer);
    isPlaying = true;

  } else {

    // Pipe the audio data to the audio player
    audioStream.pipe(audioPlayer);
    // Update the flag to indicate that the stream is playing
    isPlaying = true;
  }

}
// Kill the stream
function stopStream() {
  // Send an empty message to all connected peers
  for (const conn of conns) {
    conn.write(Buffer.alloc(0));
  }

  // Update the flag to indicate that the stream is not playing
  isPlaying = false;
}


// Generate a random public key
const publicKey = crypto.randomBytes(32);

// Create the swarm and pass in the public key
const swarm = new Hyperswarm();
gracefulGoodbye(() => swarm.destroy());

// Keep track of all connections and usernames
const conns = [];

swarm.on('connection', conn => {
  const name = b4a.toString(conn.remotePublicKey, 'hex');
  console.log(`* got a connection from ${name} (${USERNAME}) *`);

  // Start the stream if it is currently playing
  if (isPlaying) {
    startStream();
  }


  conns.push(conn);
  conn.once('close', () => conns.splice(conns.indexOf(conn), 1));
  conn.on('data', data => {
    // Check if the message is empty
    if (data.length === 0) {
      for (const conn of conns) {
        conn.write(`Stopping on all Peers`)
      }


      // Stop the stream
      audioPlayer.end()
      // Update the flag to indicate that the stream is not playing
      isPlaying = false;
    } else {
      // Pipe the data to the speaker instance
      console.log(data);

      try {
        if (!audioPlayer.writable) {
          console.log("trying")
          // Create a new audio player if it has ended
          audioPlayer = new Speaker({
            channels: 2,
            bitDepth: 16,
            sampleRate: 44100,
          });
          audioStream.pipe(audioPlayer);
          isPlaying = true;

        } else {
          audioPlayer.write(data);
        }
      } catch (err) {
        if (err.code === "ERR_STREAM_WRITE_AFTER_END") {
          console.log("The stream has already ended, cannot write data.");
        } else {
          throw err;
        }
      }
    }
  });
});


// Use readline to accept input from the user
const readline = require('readline');
const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout,
});

// When the user inputs a line of text, broadcast it to all connections
rl.on('line', input => {
  // Check if the user is issuing the !play command
  if (input.startsWith('!play')) {
    // Check if the stream is currently playing
    let dataInfo = input.split(" ")
    let URL = dataInfo[1]
    startStream(URL);
  }

  if (input === '!stop') {
    // Check if the stream is currently playing
    if (isPlaying) {
      audioPlayer.end()

      // Stop the stream
      stopStream();

      audioPlayer = new Speaker({
        channels: 2,
        bitDepth: 16,
        sampleRate: 44100,
      });
    } else {
      console.log("The stream is already stopped.");
    }
  }
});

// Join a common topic
const topic = process.argv[2] ? b4a.from(process.argv[2], 'hex') : crypto.randomBytes(32);
const discovery = swarm.join(topic, { client: true, server: true });

// The flushed promise will resolve when the topic has been fully announced to the DHT
discovery.flushed().then(() => {
  console.log(`joined topic: ${b4a.toString(topic, 'hex')}`);
  console.log("(Share this key to others so they may join)")
  console.log("To Play a youtube link, use !play LINKHERE to stop !stop")
  console.log("All commands are global to all peers")
});