echooooo.go

package main

import (
	"flag"
	"fmt"
	"github.com/bwmarrin/discordgo"
	"github.com/hraban/opus"
	"github.com/jonas747/opusutil"
	llog "log"
	"os"
	"sync"
	"time"
)

// Variables used for command line parameters
var (
	GuildID   string
	ChannelID string
	Token     string

	runningChannels = make([]chan *sync.WaitGroup, 0)
	runningLock     sync.Mutex
	Silence         = []byte{0xF8, 0xFF, 0xFE}
)

func init() {

	flag.StringVar(&GuildID, "g", "288075199415320578", "GuilID")
	flag.StringVar(&ChannelID, "c", "288079314384191488", "ChannelID")
	flag.StringVar(&Token, "t", "", "Account Token")
	flag.Parse()
}

func main() {

	llog.SetOutput(os.Stderr)

	// Create a new Discord session using the provided login information.
	// Use discordgo.New(Token) to just use a token for login.
	dg, err := discordgo.New(Token)
	if err != nil {
		log("error creating Discord session,", err)
		return
	}

	// Open the websocket and begin listening.
	err = dg.Open()
	if err != nil {
		log("error opening connection,", err)
		return
	}

	go func() {
		time.Sleep(time.Second * 5)
		log("Running")
		RunEcho(dg)
	}()

	log("Bot is now running.  Press CTRL-C to exit.")
	fmt.Scanln()
	fmt.Println("stopping")

	var wg sync.WaitGroup
	runningLock.Lock()
	for _, v := range runningChannels {
		wg.Add(1)
		v <- &wg
	}
	runningLock.Unlock()
	fmt.Println("Waiting")
	wg.Wait()
	dg.Close()

	fmt.Println("Done Waiting")
	time.Sleep(time.Second)
	return
}

// Userstream represents a individual user's audio stream.
// TODO: Optimise userstream to reuse the buffers
type UserStream struct {
	SSRC    uint32
	decoder *opus.Decoder

	bufLock sync.Mutex
	buf     []int16
}

func NewUserStream(ssrc uint32) *UserStream {
	dec, err := opus.NewDecoder(48000, 2)
	if err != nil {
		panic("Failed creating decoder: " + err.Error())
	}

	return &UserStream{
		decoder: dec,
		SSRC:    ssrc,
	}
}

// Handles an incoming voice packet
func (us *UserStream) HandlePacket(packet *discordgo.Packet) {

	header, err := opusutil.DecodeHeader(packet.Opus)
	if err != nil {
		log("Failed reading opus header: ", err)
		return
	}

	// Example: 1x 20000us frame at 48k = 1 * 20 * 48 * 2(channels) = 960 * 2 channels
	samples := int(float64(header.NumFrames)*float64(header.Config.FrameDuration.Seconds()*1000)*48) * 2
	log(packet.SSRC, ": Samples: ", samples)

	pcm := make([]int16, samples)
	_, err = us.decoder.Decode(packet.Opus, pcm)
	if err != nil {
		log("Failed deocding: ", err)
		return
		// sampleOffset += samples
		// continue
	}

	us.bufLock.Lock()
	us.buf = append(us.buf, pcm...)
	us.bufLock.Unlock()
}

func (us *UserStream) Read(b []int16) (n int, err error) {
	us.bufLock.Lock()
	if len(us.buf) < 1 {
		us.bufLock.Unlock()
		return 0, nil
	}

	n = copy(b, us.buf)
	us.buf = us.buf[n:]
	us.bufLock.Unlock()
	return
}

type Encoder struct {
	queueLock   sync.Mutex
	userStreams map[uint32]*UserStream
	stop        chan bool

	encoder *opus.Encoder

	vc *discordgo.VoiceConnection

	pcmbuf []int16
}

func NewEncoder() *Encoder {
	enc, err := opus.NewEncoder(48000, 2, opus.AppAudio)
	if err != nil {
		panic("Failed creating encoder: " + err.Error())
	}
	return &Encoder{
		stop:        make(chan bool),
		userStreams: make(map[uint32]*UserStream),
		encoder:     enc,
	}
}

func (e *Encoder) Stop() {
	close(e.stop)
}

func (e *Encoder) Queue(packet *discordgo.Packet) {

	st, ok := e.userStreams[packet.SSRC]
	if !ok {
		st = NewUserStream(packet.SSRC)
		e.queueLock.Lock()
		e.userStreams[packet.SSRC] = st
		e.queueLock.Unlock()
	}

	st.HandlePacket(packet)
}

func (e *Encoder) Run() {
	log("Encoder running")
	// ticker := time.NewTicker(time.Millisecond * 120)
	for {
		select {
		case <-e.stop:
			log("Encoder stopping")
			return
		default:
			e.ProcessQueue()
		}
	}
}

func (e *Encoder) ProcessQueue() {

	log("Processing audio")
	started := time.Now()

	e.queueLock.Lock()
	mixedPCM := make([]int16, 48*20*2)
	for _, st := range e.userStreams {

		userPCM := make([]int16, 48*20*2)
		n, _ := st.Read(userPCM)
		if n < 1 {
			continue
		}

		for i := 0; i < len(userPCM); i++ {
			// Mix it
			v := int32(mixedPCM[i] + userPCM[i])
			// Clip
			if v > 0x7fff {
				v = 0x7fff
			} else if v < -0x7fff {
				v = -0x7fff
			}
			mixedPCM[i] = int16(v)
		}
	}

	log("Took ", time.Since(started), " To process queue")
	e.queueLock.Unlock()

	output := make([]byte, 0xfff)
	n, err := e.encoder.Encode(mixedPCM, output)
	if err != nil {
		log("Failed encode: ", err)
	}
	log("Encoded ", n, " bytes")

	e.vc.OpusSend <- output[:n]
}

func RunEcho(s *discordgo.Session) {
	done := make(chan *sync.WaitGroup)
	runningLock.Lock()
	runningChannels = append(runningChannels, done)
	runningLock.Unlock()

	voice, err := s.ChannelVoiceJoin(GuildID, ChannelID, false, false)
	if err != nil {
		log("Voice err:", err)
		return
	}

	log("Waiting for voice")
	for !voice.Ready {
		time.Sleep(time.Millisecond)
	}
	log("Done waiting for voice")

	encoder := NewEncoder()
	encoder.vc = voice
	go encoder.Run()
	for {
		select {
		case packet := <-voice.OpusRecv:

			encoder.Queue(packet)
		case wg := <-done:
			voice.Close()

			encoder.Stop()

			wg.Done()
			return
		}
	}
}

func log(s ...interface{}) {
	fmt.Fprintln(os.Stderr, s...)
}