Implement a live console of the downloader on the progress page #6
@@ -20,7 +20,7 @@ func removeBOM(input []byte) []byte {
 | 
				
			|||||||
	return input
 | 
						return input
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
func downloadFile(item Item, jobInfo *JobInfo) error {
 | 
					func downloadFile(drmdFilename string, item Item, jobInfo *JobInfo) error {
 | 
				
			||||||
	logger.LogInfo("Download File", fmt.Sprintf("Starting download for: %s", item.Filename))
 | 
						logger.LogInfo("Download File", fmt.Sprintf("Starting download for: %s", item.Filename))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	tempDir := filepath.Join(config.TempBaseDir, sanitizeFilename(item.Filename))
 | 
						tempDir := filepath.Join(config.TempBaseDir, sanitizeFilename(item.Filename))
 | 
				
			||||||
@@ -131,7 +131,9 @@ func downloadFile(item Item, jobInfo *JobInfo) error {
 | 
				
			|||||||
	go func() {
 | 
						go func() {
 | 
				
			||||||
		for {
 | 
							for {
 | 
				
			||||||
			if outputBuffer.Len() > 0 {
 | 
								if outputBuffer.Len() > 0 {
 | 
				
			||||||
				broadcast(outputBuffer.Bytes())
 | 
									message := outputBuffer.Bytes()
 | 
				
			||||||
 | 
									logger.LogInfo("Download File", fmt.Sprintf("Broadcasting message: %s", message))
 | 
				
			||||||
 | 
									broadcast(drmdFilename, message)
 | 
				
			||||||
				outputBuffer.Reset()
 | 
									outputBuffer.Reset()
 | 
				
			||||||
			}
 | 
								}
 | 
				
			||||||
			time.Sleep(1 * time.Second)
 | 
								time.Sleep(1 * time.Second)
 | 
				
			||||||
 
 | 
				
			|||||||
@@ -382,19 +382,31 @@ func updateProgress(filename string, value float64, currentFile string) {
 | 
				
			|||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
var upgrader = websocket.Upgrader{}
 | 
					var upgrader = websocket.Upgrader{}
 | 
				
			||||||
var clients = make(map[*websocket.Conn]bool)
 | 
					var clients = make(map[string]map[*websocket.Conn]bool)
 | 
				
			||||||
var mu sync.Mutex
 | 
					var mu sync.Mutex
 | 
				
			||||||
 | 
					
 | 
				
			||||||
func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 | 
					func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 | 
				
			||||||
 | 
						filename := r.URL.Query().Get("filename")
 | 
				
			||||||
 | 
						fmt.Println("Filename:", filename)
 | 
				
			||||||
 | 
						if filename == "" {
 | 
				
			||||||
 | 
							http.Error(w, "Filename is required", http.StatusBadRequest)
 | 
				
			||||||
 | 
							return
 | 
				
			||||||
 | 
						}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	conn, err := upgrader.Upgrade(w, r, nil)
 | 
						conn, err := upgrader.Upgrade(w, r, nil)
 | 
				
			||||||
	if err != nil {
 | 
						if err != nil {
 | 
				
			||||||
		fmt.Println("Error while upgrading connection:", err)
 | 
							logger.LogError("WebSocket", fmt.Sprintf("Error while upgrading connection: %v", err))
 | 
				
			||||||
		return
 | 
							return
 | 
				
			||||||
	}
 | 
						}
 | 
				
			||||||
	defer conn.Close()
 | 
						defer conn.Close()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						logger.LogInfo("WebSocket", fmt.Sprintf("WebSocket connection established for filename: %s", filename))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	mu.Lock()
 | 
						mu.Lock()
 | 
				
			||||||
	clients[conn] = true
 | 
						if clients[filename] == nil {
 | 
				
			||||||
 | 
							clients[filename] = make(map[*websocket.Conn]bool)
 | 
				
			||||||
 | 
						}
 | 
				
			||||||
 | 
						clients[filename][conn] = true
 | 
				
			||||||
	mu.Unlock()
 | 
						mu.Unlock()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	for {
 | 
						for {
 | 
				
			||||||
@@ -404,18 +416,23 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
 | 
				
			|||||||
	}
 | 
						}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	mu.Lock()
 | 
						mu.Lock()
 | 
				
			||||||
	delete(clients, conn)
 | 
						delete(clients[filename], conn)
 | 
				
			||||||
	mu.Unlock()
 | 
						mu.Unlock()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						logger.LogInfo("WebSocket", fmt.Sprintf("WebSocket connection closed for filename: %s", filename))
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
func broadcast(message []byte) {
 | 
					func broadcast(filename string, message []byte) {
 | 
				
			||||||
	mu.Lock()
 | 
						mu.Lock()
 | 
				
			||||||
	defer mu.Unlock()
 | 
						defer mu.Unlock()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	for client := range clients {
 | 
						logger.LogInfo("Broadcast", fmt.Sprintf("Broadcasting message for filename: %s", filename))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						for client := range clients[filename] {
 | 
				
			||||||
		if err := client.WriteMessage(websocket.TextMessage, message); err != nil {
 | 
							if err := client.WriteMessage(websocket.TextMessage, message); err != nil {
 | 
				
			||||||
			client.Close()
 | 
								client.Close()
 | 
				
			||||||
			delete(clients, client)
 | 
								delete(clients[filename], client)
 | 
				
			||||||
 | 
								logger.LogError("Broadcast", fmt.Sprintf("Error writing message to client: %v", err))
 | 
				
			||||||
		}
 | 
							}
 | 
				
			||||||
	}
 | 
						}
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 
 | 
				
			|||||||
@@ -145,9 +145,10 @@
 | 
				
			|||||||
    <div style="display: none;" id="console"></div>
 | 
					    <div style="display: none;" id="console"></div>
 | 
				
			||||||
    <script>
 | 
					    <script>
 | 
				
			||||||
        let isPaused = false;
 | 
					        let isPaused = false;
 | 
				
			||||||
 | 
					        const filename = "{{.Filename}}";
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        function updateProgress() {
 | 
					        function updateProgress() {
 | 
				
			||||||
            fetch('/progress?filename={{.Filename}}', {
 | 
					            fetch(`/progress?filename=${filename}`, {
 | 
				
			||||||
                headers: {
 | 
					                headers: {
 | 
				
			||||||
                    'Accept': 'application/json'
 | 
					                    'Accept': 'application/json'
 | 
				
			||||||
                }
 | 
					                }
 | 
				
			||||||
@@ -179,7 +180,7 @@
 | 
				
			|||||||
        }
 | 
					        }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        function abortDownload() {
 | 
					        function abortDownload() {
 | 
				
			||||||
            fetch('/abort?filename={{.Filename}}', { method: 'POST' })
 | 
					            fetch(`/abort?filename=${filename}`, { method: 'POST' })
 | 
				
			||||||
                .then(response => {
 | 
					                .then(response => {
 | 
				
			||||||
                    if (response.ok) {
 | 
					                    if (response.ok) {
 | 
				
			||||||
                        console.log('Abort signal sent. The download will stop soon.');
 | 
					                        console.log('Abort signal sent. The download will stop soon.');
 | 
				
			||||||
@@ -190,7 +191,7 @@
 | 
				
			|||||||
        }
 | 
					        }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        function pauseDownload() {
 | 
					        function pauseDownload() {
 | 
				
			||||||
            fetch('/pause?filename={{.Filename}}', { method: 'POST' })
 | 
					            fetch(`/pause?filename=${filename}`, { method: 'POST' })
 | 
				
			||||||
                .then(response => {
 | 
					                .then(response => {
 | 
				
			||||||
                    if (response.ok) {
 | 
					                    if (response.ok) {
 | 
				
			||||||
                        console.log('Pause signal sent. The download will pause soon.');
 | 
					                        console.log('Pause signal sent. The download will pause soon.');
 | 
				
			||||||
@@ -203,7 +204,7 @@
 | 
				
			|||||||
        }
 | 
					        }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        function resumeDownload() {
 | 
					        function resumeDownload() {
 | 
				
			||||||
            fetch('/resume?filename={{.Filename}}', { method: 'POST' })
 | 
					            fetch(`/resume?filename=${filename}`, { method: 'POST' })
 | 
				
			||||||
                .then(response => {
 | 
					                .then(response => {
 | 
				
			||||||
                    if (response.ok) {
 | 
					                    if (response.ok) {
 | 
				
			||||||
                        console.log('Resume signal sent. The download will resume soon.');
 | 
					                        console.log('Resume signal sent. The download will resume soon.');
 | 
				
			||||||
@@ -217,7 +218,7 @@
 | 
				
			|||||||
        }
 | 
					        }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        const consoleDiv = document.getElementById('console');
 | 
					        const consoleDiv = document.getElementById('console');
 | 
				
			||||||
        const ws = new WebSocket(`ws://${window.location.host}/ws`);
 | 
					        const ws = new WebSocket(`ws://${window.location.host}/ws?filename=${filename}`);
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        ws.onmessage = function(event) {
 | 
					        ws.onmessage = function(event) {
 | 
				
			||||||
            consoleDiv.textContent += event.data; 
 | 
					            consoleDiv.textContent += event.data; 
 | 
				
			||||||
 
 | 
				
			|||||||
@@ -259,21 +259,25 @@ func processItems(filename string, items []Item) error {
 | 
				
			|||||||
		select {
 | 
							select {
 | 
				
			||||||
		case <-jobInfo.AbortChan:
 | 
							case <-jobInfo.AbortChan:
 | 
				
			||||||
			updateProgress(filename, 100, "Aborted")
 | 
								updateProgress(filename, 100, "Aborted")
 | 
				
			||||||
 | 
								logger.LogJobState(filename, "aborted")
 | 
				
			||||||
			return fmt.Errorf("download aborted")
 | 
								return fmt.Errorf("download aborted")
 | 
				
			||||||
		default:
 | 
							default:
 | 
				
			||||||
			if jobInfo.Paused {
 | 
								if jobInfo.Paused {
 | 
				
			||||||
				select {
 | 
									select {
 | 
				
			||||||
				case <-jobInfo.ResumeChan:
 | 
									case <-jobInfo.ResumeChan:
 | 
				
			||||||
					jobInfo.Paused = false
 | 
										jobInfo.Paused = false
 | 
				
			||||||
 | 
										logger.LogJobState(filename, "resumed")
 | 
				
			||||||
				case <-jobInfo.AbortChan:
 | 
									case <-jobInfo.AbortChan:
 | 
				
			||||||
					updateProgress(filename, 100, "Aborted")
 | 
										updateProgress(filename, 100, "Aborted")
 | 
				
			||||||
 | 
										logger.LogJobState(filename, "aborted")
 | 
				
			||||||
					return fmt.Errorf("download aborted")
 | 
										return fmt.Errorf("download aborted")
 | 
				
			||||||
				}
 | 
									}
 | 
				
			||||||
			}
 | 
								}
 | 
				
			||||||
			updateProgress(filename, float64(i)/float64(len(items))*100, items[i].Filename)
 | 
								updateProgress(filename, float64(i)/float64(len(items))*100, items[i].Filename)
 | 
				
			||||||
			err := downloadFile(items[i], jobInfo)
 | 
								err := downloadFile(filename, items[i], jobInfo)
 | 
				
			||||||
			if err != nil {
 | 
								if err != nil {
 | 
				
			||||||
				if err.Error() == "download paused" {
 | 
									if err.Error() == "download paused" {
 | 
				
			||||||
 | 
										logger.LogJobState(filename, "paused")
 | 
				
			||||||
					removeCompletedEpisodes(filename, items[:i])
 | 
										removeCompletedEpisodes(filename, items[:i])
 | 
				
			||||||
					i--
 | 
										i--
 | 
				
			||||||
					continue
 | 
										continue
 | 
				
			||||||
@@ -282,6 +286,7 @@ func processItems(filename string, items []Item) error {
 | 
				
			|||||||
		}
 | 
							}
 | 
				
			||||||
	}
 | 
						}
 | 
				
			||||||
	updateProgress(filename, 100, "")
 | 
						updateProgress(filename, 100, "")
 | 
				
			||||||
 | 
						logger.LogJobState(filename, "completed successfully")
 | 
				
			||||||
	return nil
 | 
						return nil
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 
 | 
				
			|||||||
		Reference in New Issue
	
	Block a user