Make websockets less jank, fixes #10

ambiance.go

@@ -69,20 +69,17 @@ func GetAmbiances() (amb []Ambiance, err error) {
 func AddAmbiance(uri, title string) (Ambiance, error) {
 	var amb Ambiance
 
-	msg := make(map[string]interface{})
+	ev := Event{"ambiance_add_start", map[string]string{
-	msg["event"] = "ambiance_add_start"
+		"name": title,
-	data := make(map[string]string)
+	}}
-	data["name"] = title
+
-	msg["payload"] = data
+	go ws.SendEvent(ev)
-	ws_msg <- msg
 
 	defer func() {
-		msg = make(map[string]interface{})
+		ev = Event{"ambiance_add_finish", map[string]string{
-		msg["event"] = "ambiance_add_finish"
+			"name": title,
-		data = make(map[string]string)
+		}}
-		data["name"] = title
+		go ws.SendEvent(ev)
-		msg["payload"] = data
-		ws_msg <- msg
 	}()
 
 	tmpfile, err := exec.Command("mktemp", "/tmp/dnd_XXXXXXXXXXXX.opus").Output()
@@ -134,16 +131,13 @@ func AddAmbiance(uri, title string) (Ambiance, error) {
 
 	log.Printf("Start ffmpeg to extract audio to %s", string(tmpfile))
 
-	msg = make(map[string]interface{})
+	ev = Event{"ambiance_encode_start", map[string]string{
-	msg["event"] = "ambiance_encode_start"
+		"name": title,
-	data = make(map[string]string)
+	}}
-	data["name"] = title
-	msg["payload"] = data
-	ws_msg <- msg
 
-	msg = make(map[string]interface{})
+	go ws.SendEvent(ev)
-	msg["event"] = "ambiance_encode_progress"
+
-	data = make(map[string]string)
+	data := make(map[string]string)
 	data["name"] = title
 
 	scanner := bufio.NewScanner(ffprogress)
@@ -162,8 +156,7 @@ func AddAmbiance(uri, title string) (Ambiance, error) {
 				data["percent"] = percent
 			}
 
-			msg["payload"] = data
+			go ws.SendEvent(Event{"ambiance_encode_progress", data})
-			ws_msg <- msg
 		})
 	}
 
@@ -176,15 +169,14 @@ func AddAmbiance(uri, title string) (Ambiance, error) {
 		return amb, err
 	}
 
-	msg = make(map[string]interface{})
+	ev = Event{"ambiance_encode_complete", map[string]string{
-	msg["event"] = "ambiance_encode_complete"
+		"name": title,
-	data = make(map[string]string)
+	}}
-	data["name"] = title
+
-	msg["payload"] = data
+	go ws.SendEvent(ev)
-	ws_msg <- msg
 
 	id := uuid.New()
-	fn := filepath.Join(config.GetString("ambiance.path"), fmt.Sprintf("%s.opus", id.String()))
+	fn := filepath.Join(config.GetString("ambiance.path"), fmt.Sprintf("%s.aac", id.String()))
 
 	log.Printf("Moving to %s", fn)
 

events.go

@@ -148,74 +148,70 @@ func (app *App) volset(payload ...interface{}) {
 		amb_volume.Volume = vol
 	}
 
-	app.sendVolume()
+	ev := Event{"volume", map[string]float64{
+		"playlist": pl_volume.Volume,
+		"ambiance": amb_volume.Volume,
+	}}
+
+	go ws.SendEvent(ev)
 }
 
-func (app *App) sendVolume() {
+func (app *App) songInfoEvent(event string) (ev Event, err error) {
-	msg := make(map[string]interface{})
+	ev.Event = event
-	out := make(map[string]float64)
-	msg["event"] = "volume"
-	out["playlist"] = pl_volume.Volume
-	out["ambiance"] = amb_volume.Volume
-	msg["payload"] = out
-	ws_msg <- msg
-}
 
-func (app *App) songInfoEvent(event string) map[string]interface{} {
-	msg := make(map[string]interface{})
-	msg["event"] = event
 	status, err := app.mpd.Status()
 	if err != nil {
-		log.Println(err)
+		return
-		return nil
 	}
 
 	cur, err := app.mpd.CurrentSong()
 	if err != nil {
-		log.Println(err)
+		return
-		return nil
 	}
 
 	info := new(SongInfo)
 
 	if status["state"] != "play" {
 		info.Pause = true
-		msg["payload"] = info
+		ev.Payload = info
-		return msg
+		return
 	}
 
 	duration, ok := status["duration"]
 	if ok && duration != "" {
-		slen, err := strconv.ParseFloat(duration, 64)
+		var slen float64
+		slen, err = strconv.ParseFloat(duration, 64)
 		if err != nil {
 			log.Println(err)
-			return nil
+			return
 		}
 		info.Length = time.Duration(slen * float64(time.Second)).Milliseconds()
 	}
 
 	elapsed, ok := status["elapsed"]
 	if ok && elapsed != "" {
-		spos, err := strconv.ParseFloat(elapsed, 64)
+		var spos float64
+		spos, err = strconv.ParseFloat(elapsed, 64)
 		if err != nil {
 			log.Println(err)
-			return nil
+			return
 		}
 		info.Position = time.Duration(spos * float64(time.Second)).Milliseconds()
 	}
 
 	album, ok := cur["Album"]
 	if ok {
-		plid, err := uuid.ParseBytes([]byte(album))
+		var plid uuid.UUID
+		plid, err = uuid.ParseBytes([]byte(album))
 		if err != nil {
 			log.Println(err)
-			return nil
+			return
 		}
-
+		var pl *Playlist
-		pl, err := app.GetPlaylist(plid)
+		pl, err = app.GetPlaylist(plid)
 		if err != nil {
 			log.Println(err)
-			return nil
+			return
 		}
 
 		info.Playlist = pl.Id
@@ -237,9 +233,9 @@ func (app *App) songInfoEvent(event string) map[string]interface{} {
 		info.Song = location
 	}
 
-	msg["payload"] = *info
+	ev.Payload = *info
 
-	return msg
+	return ev, nil
 }
 
 func (app *App) ambiancePlay(payload ...interface{}) {
@@ -292,15 +288,13 @@ func (app *App) ambiancePlay(payload ...interface{}) {
 		log.Fatal(err)
 	}
 
-	msg := make(map[string]interface{})
-	out := make(map[string]interface{})
-
 	app.curamb = amb
 
-	msg["event"] = "ambiance_play"
+	ev := Event{"ambiance_play", map[string]string{
-	out["id"] = id
+		"id": id,
-	msg["payload"] = out
+	}}
-	ws_msg <- msg
+
+	go ws.SendEvent(ev)
 }
 
 func (app *App) ambianceStop(payload ...interface{}) {
@@ -319,10 +313,7 @@ func (app *App) ambianceStop(payload ...interface{}) {
 		log.Fatal(err)
 	}
 
-	msg := make(map[string]interface{})
+	go ws.SendEvent(Event{"ambiance_stop", nil})
-	msg["event"] = "ambiance_stop"
-	ws_msg <- msg
-
 }
 
 func (app *App) ambianceAdd(payload ...interface{}) {
@@ -359,13 +350,12 @@ func (app *App) ambianceAdd(payload ...interface{}) {
 		return
 	}
 
-	msg := make(map[string]interface{})
+	ev := Event{"ambiance_add", map[string]string{
-	out := make(map[string]interface{})
+		"title": amb.Title,
-	msg["event"] = "ambiance_add"
+		"id":    amb.Id,
-	out["title"] = amb.Title
+	}}
-	out["id"] = amb.Id
+
-	msg["payload"] = out
+	go ws.SendEvent(ev)
-	ws_msg <- msg
 }
 
 func (app *App) songPosition(payload ...interface{}) {
@@ -380,18 +370,16 @@ func (app *App) songPosition(payload ...interface{}) {
 	}
 
 	l.Do(func() {
-		msg := make(map[string]interface{})
-		out := make(map[string]interface{})
 
 		slen, _ := strconv.ParseFloat(status["duration"], 64)
 		spos, _ := strconv.ParseFloat(status["elapsed"], 64)
 
-		msg["event"] = "song_position"
+		ev := Event{"song_position", map[string]int64{
-		out["len"] = time.Duration(slen * float64(time.Second)).Milliseconds()
+			"len":      time.Duration(slen * float64(time.Second)).Milliseconds(),
-		out["position"] = time.Duration(spos * float64(time.Second)).Milliseconds()
+			"position": time.Duration(spos * float64(time.Second)).Milliseconds(),
+		}}
 
-		msg["payload"] = out
+		go ws.SendEvent(ev)
-		ws_msg <- msg
 	})
 
 }
@@ -399,10 +387,12 @@ func (app *App) songPosition(payload ...interface{}) {
 func (app *App) songInfo(payload ...interface{}) {
 	log.Println("song_info event received")
 
-	msg := app.songInfoEvent("song_info")
+	ev, err := app.songInfoEvent("song_info")
-	if msg != nil {
+	if err != nil {
-		ws_msg <- msg
+		log.Println(err)
+		return
 	}
+	go ws.SendEvent(ev)
 }
 
 func (app *App) stop(payload ...interface{}) {
@@ -415,10 +405,7 @@ func (app *App) stop(payload ...interface{}) {
 	app.mpd.Stop()
 	app.plmutex.Unlock()
 
-	msg := make(map[string]interface{})
+	go ws.SendEvent(Event{"stop", nil})
-	msg["event"] = "stop"
-
-	ws_msg <- msg
 }
 
 func (app *App) prevSong(payload ...interface{}) {
@@ -492,12 +479,12 @@ func (app *App) addPlaylist(payload ...interface{}) {
 		log.Println("Error getting youtube playlist info,", plid)
 	}
 
-	msg := make(map[string]interface{})
+	ev := Event{"new_playlist", map[string]string{
+		"url":   id.String(),
+		"title": pltitle,
+	}}
 
-	msg["event"] = "new_playlist"
+	go ws.SendEvent(ev)
-	msg["payload"] = map[string]string{"url": id.String(), "title": pltitle}
-
-	ws_msg <- msg
 }
 
 func (app *App) loadPlaylist(payload ...interface{}) {

routes.go

@@ -69,7 +69,7 @@ func init() {
 			return
 		}
 
-		err = handleWS(conn)
+		err = ws.join(conn)
 		if err != nil {
 			log.Printf("WS connection closed, %v\n", r.RemoteAddr)
 		}

ws.go

@@ -12,18 +12,21 @@ import (
 	"github.com/kataras/go-events"
 )
 
+type Websocket struct {
+	sync.Mutex
+
+	clients *bcast.Group
+}
+
+var ws *Websocket
+
 func init() {
 	log.Println("ws.go loading..")
-	go ws_clients.Broadcast(0)
+	ws = new(Websocket)
-	ws_msg = make(chan interface{})
 
-	go func() {
+	ws.clients = bcast.NewGroup()
-		var msg interface{}
+
-		for {
+	go ws.clients.Broadcast(0)
-			msg = <-ws_msg
-			ws_clients.Send(msg)
-		}
-	}()
 
 	log.Println("ws.go done.")
 }
@@ -33,12 +36,19 @@ type WSmsg struct {
 	Payload json.RawMessage
 }
 
-var ws_clients = bcast.NewGroup()
+type Event struct {
-var ws_msg chan interface{}
+	Event   string `json:"event"`
-var WSMutex = &sync.Mutex{}
+	Payload any    `json:"payload,omitempty"`
+}
 
-func handleWS(c *websocket.Conn) error {
+func (ws *Websocket) SendEvent(e Event) {
-	memb := ws_clients.Join()
+	ws.Lock()
+	ws.clients.Send(e)
+	ws.Unlock()
+}
+
+func (ws *Websocket) join(c *websocket.Conn) error {
+	memb := ws.clients.Join()
 	defer memb.Close()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -64,29 +74,27 @@ func handleWS(c *websocket.Conn) error {
 		return nil
 	})
 
-	msg := app.songInfoEvent("song_info")
+	msg, err := app.songInfoEvent("song_info")
+	if err != nil {
+		return err
+	}
 
-	vol := make(map[string]interface{})
+	vol := Event{"volume", map[string]float64{
-	volout := make(map[string]float64)
+		"playlist": pl_volume.Volume,
-	vol["event"] = "volume"
+		"ambiance": amb_volume.Volume,
-	volout["playlist"] = pl_volume.Volume
+	}}
-	volout["ambiance"] = amb_volume.Volume
-	vol["payload"] = volout
 
 	c.SetWriteDeadline(time.Now().Add(10 * time.Second))
 	c.WriteJSON(msg)
 	c.WriteJSON(vol)
 
 	if app.ambiance.Len() > 0 {
-		msg := make(map[string]interface{})
+		msg := Event{"ambiance_play", map[string]string{
-		out := make(map[string]interface{})
+			"id": app.curamb.Id,
-		msg["event"] = "ambiance_play"
+		}}
-		out["id"] = app.curamb.Id
-		msg["payload"] = out
 		c.WriteJSON(msg)
 	} else {
-		msg := make(map[string]interface{})
+		msg := Event{"ambiance_stop", nil}
-		msg["event"] = "ambiance_stop"
 		c.WriteJSON(msg)
 	}
 
@@ -97,6 +105,6 @@ func handleWS(c *websocket.Conn) error {
 			return err
 		}
 
-		app.events.Emit(events.EventName(msg.Event), msg.Payload)
+		app.events.Emit(events.EventName(msg.Event), msg.Payload, memb)
 	}
 }

ytdl.go

@@ -1,23 +1,42 @@
 package main
 
 import (
-	"bufio"
-	"encoding/json"
 	"fmt"
-	"log"
-	"math"
 	"net/url"
-	"os"
-	"os/exec"
-	"strconv"
 	"time"
 
 	"golang.org/x/time/rate"
 )
 
 var prate = rate.Sometimes{Interval: 1 * time.Second}
-var yturl = "https://youtu.be/%s"
 
+// var yturl = "https://youtu.be/%s"
+
+func YTUrl(uri string) (vid string, err error) {
+	u, err := url.Parse(uri)
+	if err != nil {
+		return "", err
+	}
+
+	switch u.Host {
+	case "youtu.be":
+		vid = u.Path[1:]
+	case "m.youtube.com":
+		fallthrough
+	case "youtube.com":
+		fallthrough
+	case "www.youtube.com":
+		vid = u.Query().Get("v")
+	}
+
+	if vid == "" {
+		return vid, fmt.Errorf("unable to parse vid")
+	}
+
+	return
+}
+
+/*
 func NewYTdlUrl(vid string) ([]byte, error) {
 	ytdl := config.GetString("youtube.ytdl")
 	yt := exec.Command(
@@ -132,29 +151,6 @@ func NewYTdl(vid string) ([]byte, error) {
 	return tmpfile, nil
 }
 
-func YTUrl(uri string) (vid string, err error) {
-	u, err := url.Parse(uri)
-	if err != nil {
-		return "", err
-	}
-
-	switch u.Host {
-	case "youtu.be":
-		vid = u.Path[1:]
-	case "m.youtube.com":
-		fallthrough
-	case "youtube.com":
-		fallthrough
-	case "www.youtube.com":
-		vid = u.Query().Get("v")
-	}
-
-	if vid == "" {
-		return vid, fmt.Errorf("unable to parse vid")
-	}
-
-	return
-}
 
 /*
 func DownloadAmbiance(uri string, name string) error {