implement websocket to send reindex messages

apiGo/videoparser/Helpers.go

@@ -0,0 +1,120 @@
+package videoparser
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"os/exec"
+	"strconv"
+)
+
+func AppendMessage(message string) {
+	msger := TextMessage{
+		MessageBase: MessageBase{Action: "message"},
+		Message:     message,
+	}
+	marshal, err := json.Marshal(msger)
+	if err != nil {
+		return
+	}
+
+	IndexSender.Publish(marshal)
+}
+
+func SendEvent(message string) {
+	msger := ReindexEvent{
+		MessageBase: MessageBase{Action: "reindexAction"},
+		Event:       message,
+	}
+	marshal, err := json.Marshal(msger)
+	if err != nil {
+		return
+	}
+
+	IndexSender.Publish(marshal)
+}
+
+// ext dependency support check
+func checkExtDependencySupport() *ExtDependencySupport {
+	var extDepsAvailable ExtDependencySupport
+
+	extDepsAvailable.FFMpeg = commandExists("ffmpeg")
+	extDepsAvailable.MediaInfo = commandExists("mediainfo")
+
+	return &extDepsAvailable
+}
+
+// check if a specific system command is available
+func commandExists(cmd string) bool {
+	_, err := exec.LookPath(cmd)
+	return err == nil
+}
+
+// parse the thumbail picture from video file
+func parseFFmpegPic(path string) (*string, error) {
+	app := "ffmpeg"
+
+	cmd := exec.Command(app,
+		"-hide_banner",
+		"-loglevel", "panic",
+		"-ss", "00:04:00",
+		"-i", path,
+		"-vframes", "1",
+		"-q:v", "2",
+		"-f", "singlejpeg",
+		"pipe:1")
+
+	stdout, err := cmd.Output()
+
+	if err != nil {
+		fmt.Println(err.Error())
+		fmt.Println(string(err.(*exec.ExitError).Stderr))
+		return nil, err
+	}
+
+	strEncPic := base64.StdEncoding.EncodeToString(stdout)
+	if strEncPic == "" {
+		return nil, nil
+	}
+	backpic64 := fmt.Sprintf("data:image/jpeg;base64,%s", strEncPic)
+
+	return &backpic64, nil
+}
+
+func getVideoAttributes(path string) *VideoAttributes {
+	app := "mediainfo"
+
+	arg0 := path
+	arg1 := "--Output=JSON"
+
+	cmd := exec.Command(app, arg1, "-f", arg0)
+	stdout, err := cmd.Output()
+
+	var t struct {
+		Media struct {
+			Track []struct {
+				Duration string
+				FileSize string
+				Width    string
+			}
+		}
+	}
+	err = json.Unmarshal(stdout, &t)
+
+	if err != nil {
+		fmt.Println(err.Error())
+		return nil
+	}
+
+	duration, err := strconv.ParseFloat(t.Media.Track[0].Duration, 32)
+	filesize, err := strconv.Atoi(t.Media.Track[0].FileSize)
+	width, err := strconv.Atoi(t.Media.Track[1].Width)
+
+	ret := VideoAttributes{
+		Duration: float32(duration),
+		FileSize: uint(filesize),
+		Width:    uint(width),
+	}
+
+	return &ret
+}

apiGo/videoparser/ReIndexTVShows.go

@@ -0,0 +1,7 @@
+package videoparser
+
+import "openmediacenter/apiGo/api/types"
+
+func startTVShowReindex(files []Show, sett types.SettingsType) {
+	// have fun with db insertions here!
+}

apiGo/videoparser/ReIndexVideos.go

@@ -2,13 +2,10 @@ package videoparser
 
 import (
 	"database/sql"
-	"encoding/base64"
-	"encoding/json"
 	"fmt"
 	"openmediacenter/apiGo/api/types"
 	"openmediacenter/apiGo/database"
 	"openmediacenter/apiGo/videoparser/tmdb"
-	"os/exec"
 	"regexp"
 	"strconv"
 	"strings"
@@ -51,9 +48,8 @@ func ReIndexVideos(path []string, sett types.SettingsType) {
 		processVideo(s)
 	}
 
-	AppendMessageBuffer("reindex finished successfully!")
-
-	contentAvailable = false
+	AppendMessage("reindex finished successfully!")
+	SendEvent("stop")
 	fmt.Println("Reindexing finished!")
 }
 
@@ -125,7 +121,7 @@ func addVideo(videoName string, fileName string, year int) {
 	}
 
 	if mExtDepsAvailable.FFMpeg {
-		ppic, err = parseFFmpegPic(fileName)
+		ppic, err = parseFFmpegPic(mSettings.VideoPath + fileName)
 		if err != nil {
 			fmt.Printf("FFmpeg error occured: %s\n", err.Error())
 		} else {
@@ -134,7 +130,7 @@ func addVideo(videoName string, fileName string, year int) {
 	}
 
 	if mExtDepsAvailable.MediaInfo {
-		atr := getVideoAttributes(fileName)
+		atr := getVideoAttributes(mSettings.VideoPath + fileName)
 		if atr != nil {
 			vidAtr = atr
 		}
@@ -168,7 +164,7 @@ func addVideo(videoName string, fileName string, year int) {
 		insertTMDBTags(tmdbData.GenreIds, insertId)
 	}
 
-	AppendMessageBuffer(fmt.Sprintf("%s - added!", videoName))
+	AppendMessage(fmt.Sprintf("%s - added!", videoName))
 }
 
 func matchYear(fileName string) (int, string) {
@@ -189,95 +185,6 @@ func matchYear(fileName string) (int, string) {
 	return year, r.ReplaceAllString(fileName, "")
 }
 
-// parse the thumbail picture from video file
-func parseFFmpegPic(fileName string) (*string, error) {
-	app := "ffmpeg"
-
-	cmd := exec.Command(app,
-		"-hide_banner",
-		"-loglevel", "panic",
-		"-ss", "00:04:00",
-		"-i", mSettings.VideoPath+fileName,
-		"-vframes", "1",
-		"-q:v", "2",
-		"-f", "singlejpeg",
-		"pipe:1")
-
-	stdout, err := cmd.Output()
-
-	if err != nil {
-		fmt.Println(err.Error())
-		fmt.Println(string(err.(*exec.ExitError).Stderr))
-		return nil, err
-	}
-
-	strEncPic := base64.StdEncoding.EncodeToString(stdout)
-	if strEncPic == "" {
-		return nil, nil
-	}
-	backpic64 := fmt.Sprintf("data:image/jpeg;base64,%s", strEncPic)
-
-	return &backpic64, nil
-}
-
-func getVideoAttributes(fileName string) *VideoAttributes {
-	app := "mediainfo"
-
-	arg0 := mSettings.VideoPath + fileName
-	arg1 := "--Output=JSON"
-
-	cmd := exec.Command(app, arg1, "-f", arg0)
-	stdout, err := cmd.Output()
-
-	var t struct {
-		Media struct {
-			Track []struct {
-				Duration string
-				FileSize string
-				Width    string
-			}
-		}
-	}
-	err = json.Unmarshal(stdout, &t)
-
-	if err != nil {
-		fmt.Println(err.Error())
-		return nil
-	}
-
-	duration, err := strconv.ParseFloat(t.Media.Track[0].Duration, 32)
-	filesize, err := strconv.Atoi(t.Media.Track[0].FileSize)
-	width, err := strconv.Atoi(t.Media.Track[1].Width)
-
-	ret := VideoAttributes{
-		Duration: float32(duration),
-		FileSize: uint(filesize),
-		Width:    uint(width),
-	}
-
-	return &ret
-}
-
-func AppendMessageBuffer(message string) {
-	messageBuffer = append(messageBuffer, message)
-}
-
-// ext dependency support check
-func checkExtDependencySupport() *ExtDependencySupport {
-	var extDepsAvailable ExtDependencySupport
-
-	extDepsAvailable.FFMpeg = commandExists("ffmpeg")
-	extDepsAvailable.MediaInfo = commandExists("mediainfo")
-
-	return &extDepsAvailable
-}
-
-// check if a specific system command is available
-func commandExists(cmd string) bool {
-	_, err := exec.LookPath(cmd)
-	return err == nil
-}
-
 // insert the default size tags to corresponding video
 func insertSizeTag(width uint, videoId uint) {
 	var tagType uint

apiGo/videoparser/VideoParser.go

@@ -2,25 +2,22 @@ package videoparser
 
 import (
 	"fmt"
+	"io/ioutil"
 	"openmediacenter/apiGo/database"
 	"os"
 	"path/filepath"
 	"strings"
 )
 
-var messageBuffer []string
-var contentAvailable = false
-
 type StatusMessage struct {
 	Messages         []string
 	ContentAvailable bool
 }
 
 func StartReindex() bool {
-	messageBuffer = []string{}
-	contentAvailable = true
-
 	fmt.Println("starting reindex..")
+	SendEvent("start")
+	AppendMessage("starting reindex..")
 
 	mSettings := database.GetSettings()
 	// add the path prefix to videopath
@@ -29,6 +26,8 @@ func StartReindex() bool {
 	// check if path even exists
 	if _, err := os.Stat(mSettings.VideoPath); os.IsNotExist(err) {
 		fmt.Println("Reindex path doesn't exist!")
+		AppendMessage(fmt.Sprintf("Reindex path doesn't exist! :%s", mSettings.VideoPath))
+		SendEvent("stop")
 		return false
 	}
 
@@ -49,25 +48,79 @@ func StartReindex() bool {
 		fmt.Println(err.Error())
 	}
 	// start reindex process
-	AppendMessageBuffer("Starting Reindexing!")
+	AppendMessage("Starting Reindexing!")
 	go ReIndexVideos(files, mSettings)
 	return true
 }
 
-// StartTVShowReindex reindex dir walks for TVShow reindex
-func StartTVShowReindex() {
-	// todo implement walking through dirs and reindex!
+type Show struct {
+	Name  string
+	files []string
 }
 
-func GetStatusMessage() *StatusMessage {
-	msg := StatusMessage{
-		Messages:         messageBuffer,
-		ContentAvailable: contentAvailable,
+// StartTVShowReindex reindex dir walks for TVShow reindex
+func StartTVShowReindex() {
+	fmt.Println("starting tvshow reindex..")
+	SendEvent("start")
+	AppendMessage("starting tvshow reindex...")
+
+	mSettings := database.GetSettings()
+	// add the path prefix to videopath
+	mSettings.EpisodePath = mSettings.PathPrefix + mSettings.EpisodePath
+
+	// add slash suffix if not existing
+	if !strings.HasSuffix(mSettings.EpisodePath, "/") {
+		mSettings.EpisodePath += "/"
 	}
 
-	messageBuffer = []string{}
+	// check if path even exists
+	if _, err := os.Stat(mSettings.EpisodePath); os.IsNotExist(err) {
+		msg := fmt.Sprintf("Reindex path doesn't exist! :%s", mSettings.EpisodePath)
+		fmt.Println(msg)
+		AppendMessage(msg)
+		SendEvent("stop")
+		return
+	}
 
-	return &msg
+	var files []Show
+
+	filess, err := ioutil.ReadDir(mSettings.EpisodePath)
+	if err != nil {
+		fmt.Println(err.Error())
+	}
+
+	for _, file := range filess {
+		if file.IsDir() {
+			elem := Show{
+				Name:  file.Name(),
+				files: nil,
+			}
+
+			fmt.Println(file.Name())
+
+			episodefiles, err := ioutil.ReadDir(mSettings.EpisodePath + file.Name())
+			if err != nil {
+				fmt.Println(err.Error())
+			}
+
+			for _, epfile := range episodefiles {
+				if strings.HasSuffix(epfile.Name(), ".mp4") {
+					elem.files = append(elem.files, epfile.Name())
+				}
+			}
+			files = append(files, elem)
+		}
+	}
+
+	fmt.Println(files)
+
+	if err != nil {
+		fmt.Println(err.Error())
+	}
+
+	// start reindex process
+	AppendMessage("Starting Reindexing!")
+	go startTVShowReindex(files, mSettings)
 }
 
 func StartCleanup() {

apiGo/videoparser/WebSocketConnector.go

@@ -0,0 +1,139 @@
+package videoparser
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"nhooyr.io/websocket"
+	"sync"
+	"time"
+)
+
+// subscriber represents a subscriber.
+// Messages are sent on the msgs channel and if the client
+// cannot keep up with the messages, closeSlow is called.
+type subscriber struct {
+	msgs      chan []byte
+	closeSlow func()
+}
+
+type ChatSender struct {
+	subscribersMu sync.Mutex
+	subscribers   map[*subscriber]struct{}
+}
+
+func newChatSender() *ChatSender {
+	return &ChatSender{
+		subscribers: make(map[*subscriber]struct{}),
+	}
+}
+
+func (t *ChatSender) TestCall() {
+	fmt.Println("hello world")
+}
+
+func (t *ChatSender) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
+		OriginPatterns: []string{"*"},
+	})
+	if err != nil {
+		fmt.Println(err.Error())
+		return
+	}
+	defer c.Close(websocket.StatusInternalError, "")
+
+	err = t.subscribe(r.Context(), c)
+	if errors.Is(err, context.Canceled) {
+		return
+	}
+	if websocket.CloseStatus(err) == websocket.StatusNormalClosure ||
+		websocket.CloseStatus(err) == websocket.StatusGoingAway {
+		return
+	}
+	if err != nil {
+		fmt.Println(err.Error())
+		return
+	}
+}
+
+func (t *ChatSender) subscribe(ctx context.Context, c *websocket.Conn) error {
+	ctx = c.CloseRead(ctx)
+
+	s := &subscriber{
+		msgs: make(chan []byte, 16),
+		closeSlow: func() {
+			c.Close(websocket.StatusPolicyViolation, "connection too slow to keep up with messages")
+		},
+	}
+	t.addSubscriber(s)
+	defer t.deleteSubscriber(s)
+
+	for {
+		select {
+		case msg := <-s.msgs:
+			err := writeTimeout(ctx, time.Second*5, c, msg)
+			if err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+type MessageBase struct {
+	Action string
+}
+
+type TextMessage struct {
+	MessageBase
+
+	Message string
+}
+
+type ReindexEvent struct {
+	MessageBase
+
+	Event string
+}
+
+func (t *ChatSender) Publish(msg []byte) {
+	t.subscribersMu.Lock()
+	defer t.subscribersMu.Unlock()
+
+	for s := range t.subscribers {
+		select {
+		case s.msgs <- msg:
+		default:
+			go s.closeSlow()
+		}
+	}
+}
+
+var IndexSender = newChatSender()
+
+func SetupSettingsWebsocket() {
+	http.Handle("/subscribe", IndexSender)
+}
+
+// addSubscriber registers a subscriber.
+func (t *ChatSender) addSubscriber(s *subscriber) {
+	t.subscribersMu.Lock()
+	t.subscribers[s] = struct{}{}
+	t.subscribersMu.Unlock()
+}
+
+// deleteSubscriber deletes the given subscriber.
+func (t *ChatSender) deleteSubscriber(s *subscriber) {
+	t.subscribersMu.Lock()
+	delete(t.subscribers, s)
+	t.subscribersMu.Unlock()
+}
+
+func writeTimeout(ctx context.Context, timeout time.Duration, c *websocket.Conn, msg []byte) error {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	return c.Write(ctx, websocket.MessageText, msg)
+}