hpz937/zurg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Remove ristretto cache

Browse Source
This commit is contained in:
Ben Sarmiento
2023-12-06 01:11:58 +01:00
parent 121a28d46f
commit 4b8fd82acd
7 changed files with 63 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
internal/torrent/manager.go
View File

@@ -4,18 +4,14 @@ import (
"fmt"
"io"
"math"
"net/url"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
"github.com/debridmediamanager/zurg/internal/config"
"github.com/debridmediamanager/zurg/pkg/dav"
"github.com/debridmediamanager/zurg/pkg/realdebrid"
"github.com/dgraph-io/ristretto"
cmap "github.com/orcaman/concurrent-map/v2"
"github.com/panjf2000/ants/v2"
"github.com/scylladb/go-set"
@@ -33,7 +29,6 @@ type TorrentManager struct {
Api *realdebrid.RealDebrid
DirectoryMap cmap.ConcurrentMap[string, cmap.ConcurrentMap[string, *Torrent]] // directory -> accessKey -> Torrent
DownloadCache cmap.ConcurrentMap[string, *realdebrid.Download]
ResponseCache *ristretto.Cache
accessKeySet *strset.Set
latestState *LibraryState
requiredVersion string
@@ -45,13 +40,12 @@ type TorrentManager struct {
// NewTorrentManager creates a new torrent manager
// it will fetch all torrents and their info in the background
// and store them in-memory and cached in files
func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, p *ants.Pool, cache *ristretto.Cache, log *zap.SugaredLogger) *TorrentManager {
func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, p *ants.Pool, log *zap.SugaredLogger) *TorrentManager {
initialSate := EmptyState()
t := &TorrentManager{
Config: cfg,
Api: api,
ResponseCache: cache,
accessKeySet: set.NewStringSet(),
latestState: &initialSate,
requiredVersion: "03.12.2023",
@@ -116,11 +110,11 @@ func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, p
return t
}
func (t *TorrentManager) RefreshTorrents() []string {
func (t *TorrentManager) RefreshTorrents() {
instances, _, err := t.Api.GetTorrents(0)
if err != nil {
t.log.Warnf("Cannot get torrents: %v\n", err)
return nil
return
}
infoChan := make(chan *Torrent, len(instances))
var wg sync.WaitGroup
@@ -154,32 +148,18 @@ func (t *TorrentManager) RefreshTorrents() []string {
}
t.log.Infof("Compiled into %d torrents, %d were missing info", oldTorrents.Count(), noInfoCount)
var updatedPaths []string
somethingChanged := false
// removed
strset.Difference(t.accessKeySet, freshKeys).Each(func(accessKey string) bool {
somethingChanged = true
torrentPaths := t.Delete(accessKey, false, false)
updatedPaths = append(updatedPaths, torrentPaths...)
t.Delete(accessKey, false)
return true
})
// new
strset.Difference(freshKeys, t.accessKeySet).Each(func(accessKey string) bool {
somethingChanged = true
torrent, _ := oldTorrents.Get(accessKey)
torrentPaths := t.UpdateTorrentResponseCache(torrent)
updatedPaths = append(updatedPaths, torrentPaths...)
t.accessKeySet.Add(accessKey)
return true
})
// now we can build the directory responses
if somethingChanged {
t.UpdateDirectoryResponsesCache()
}
t.SetNewLatestState(t.getCurrentState())
return updatedPaths
}
// getMoreInfo gets original name, size and files for a torrent
@@ -400,13 +380,9 @@ func (t *TorrentManager) startRefreshJob() {
}
t.log.Infof("Detected changes! Refreshing %d torrents", checksum.TotalCount)
updatedPaths := t.RefreshTorrents()
t.RefreshTorrents()
t.log.Info("Finished refreshing torrents")
if updatedPaths != nil {
t.TriggerHookOnLibraryUpdate(updatedPaths)
}
if t.Config.EnableRepair() {
t.RepairAll()
} else {
@@ -539,7 +515,7 @@ func (t *TorrentManager) CheckDeletedState(torrent *Torrent) bool {
return false
}
func (t *TorrentManager) Delete(accessKey string, deleteInRD bool, updateDirectoryResponses bool) []string {
func (t *TorrentManager) Delete(accessKey string, deleteInRD bool) {
if deleteInRD {
allTorrents, _ := t.DirectoryMap.Get(INT_ALL)
infoCache, _ := t.DirectoryMap.Get(INT_INFO_CACHE)
@@ -554,19 +530,9 @@ func (t *TorrentManager) Delete(accessKey string, deleteInRD bool, updateDirecto
}
}
t.log.Infof("Removing torrent %s from zurg database", accessKey)
var updatedPaths []string
t.DirectoryMap.IterCb(func(directory string, torrents cmap.ConcurrentMap[string, *Torrent]) {
if ok := torrents.Has(accessKey); ok {
torrents.Remove(accessKey)
pathKey := fmt.Sprintf("%s/%s", directory, accessKey)
updatedPaths = append(updatedPaths, pathKey)
t.ResponseCache.Del(pathKey)
}
torrents.Remove(accessKey)
})
if updateDirectoryResponses {
t.UpdateDirectoryResponsesCache()
}
return updatedPaths
}
func (t *TorrentManager) repair(torrent *Torrent) {
@@ -744,77 +710,6 @@ func (t *TorrentManager) Repair(torrent *Torrent) {
t.repair(torrent)
t.log.Info("Finished repairing torrent %s", torrent.AccessKey)
})
updatedPaths := t.UpdateTorrentResponseCache(torrent)
t.TriggerHookOnLibraryUpdate(updatedPaths)
}
func (t *TorrentManager) UpdateTorrentResponseCache(torrent *Torrent) []string {
updatedPaths := []string{}
dav, html := t.buildTorrentResponses(torrent)
t.AssignedDirectoryCb(torrent, func(directory string) {
if strings.HasPrefix(directory, "int__") {
return
}
torrents, _ := t.DirectoryMap.Get(directory)
torrents.Set(torrent.AccessKey, torrent)
// torrent responses
pathKey := fmt.Sprintf("%s/%s", directory, torrent.AccessKey)
updatedPaths = append(updatedPaths, pathKey)
newHtml := strings.ReplaceAll(html, "$dir", directory)
t.ResponseCache.Set(pathKey+".html", newHtml, 1)
newDav := strings.ReplaceAll(dav, "$dir", directory)
t.ResponseCache.Set(pathKey+".dav", newDav, 1)
})
return updatedPaths
}
func (t *TorrentManager) UpdateDirectoryResponsesCache() {
t.DirectoryMap.IterCb(func(directory string, torrents cmap.ConcurrentMap[string, *Torrent]) {
allKeys := torrents.Keys()
sort.Strings(allKeys)
davRet := ""
htmlRet := ""
for _, accessKey := range allKeys {
if tor, ok := torrents.Get(accessKey); ok {
if tor.AllInProgress() {
continue
}
davRet += dav.Directory(tor.AccessKey, tor.LatestAdded)
htmlRet += fmt.Sprintf("<li><a href=\"/http/%s/%s/\">%s</a></li>", directory, tor.AccessKey, tor.AccessKey)
}
}
cacheKey := directory
davRet = "<?xml version=\"1.0\" encoding=\"utf-8\"?><d:multistatus xmlns:d=\"DAV:\">" + dav.BaseDirectory(directory, "") + dav.BaseDirectory(directory, "") + davRet + "</d:multistatus>"
t.ResponseCache.Set(cacheKey+".dav", davRet, 1)
htmlRet = "<ol>" + htmlRet
t.ResponseCache.Set(cacheKey+".html", htmlRet, 1)
})
}
func (t *TorrentManager) buildTorrentResponses(tor *Torrent) (string, string) {
davRet := "<?xml version=\"1.0\" encoding=\"utf-8\"?><d:multistatus xmlns:d=\"DAV:\">" + dav.BaseDirectory(filepath.Join("$dir", tor.AccessKey), tor.LatestAdded)
htmlRet := "<ol>"
filenames := tor.SelectedFiles.Keys()
sort.Strings(filenames)
for _, filename := range filenames {
file, _ := tor.SelectedFiles.Get(filename)
if file == nil || !strings.HasPrefix(file.Link, "http") {
// will be caught by torrent manager's repairAll
// just skip it for now
continue
}
davRet += dav.File(filename, file.Bytes, file.Ended)
filePath := filepath.Join("$dir", tor.AccessKey, url.PathEscape(filename))
htmlRet += fmt.Sprintf("<li><a href=\"/http/%s\">%s</a></li>", filePath, filename)
}
davRet += "</d:multistatus>"
return davRet, htmlRet
}
func (t *TorrentManager) AssignedDirectoryCb(tor *Torrent, cb func(string)) {