deranjer/goTorrent
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

starting to redo the core to do queuing and better downloading

Browse Source
This commit is contained in:
deranjer
2018-04-18 22:41:32 -04:00
parent 0fdc926cc4
commit 4909429390
9 changed files with 136 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
engine/cronJobs.go
View File

@@ -22,7 +22,7 @@ func InitializeCronEngine() *cron.Cron {
}
//CheckTorrentWatchFolder adds torrents from a watch folder //TODO see if you can use filepath.Abs instead of changing directory
func CheckTorrentWatchFolder(c *cron.Cron, db *storm.DB, tclient *torrent.Client, torrentLocalStorage Storage.TorrentLocal, config Settings.FullClientSettings) {
func CheckTorrentWatchFolder(c *cron.Cron, db *storm.DB, tclient *torrent.Client, torrentLocalStorage Storage.TorrentLocal, config Settings.FullClientSettings, activeTorrents []string, queuedTorrents []string) {
c.AddFunc("@every 5m", func() {
Logger.WithFields(logrus.Fields{"Watch Folder": config.TorrentWatchFolder}).Info("Running the watch folder cron job")
torrentFiles, err := ioutil.ReadDir(config.TorrentWatchFolder)
@@ -50,7 +50,7 @@ func CheckTorrentWatchFolder(c *cron.Cron, db *storm.DB, tclient *torrent.Client
os.Remove(fullFilePathAbs) //delete the torrent after adding it and copying it over
Logger.WithFields(logrus.Fields{"Source Folder": fullFilePathAbs, "Destination Folder": fullNewFilePathAbs, "Torrent": file.Name()}).Info("Added torrent from watch folder, and moved torrent file")
StartTorrent(clientTorrent, torrentLocalStorage, db, "file", fullNewFilePathAbs, config.DefaultMoveFolder, "default", config)
AddTorrent(clientTorrent, torrentLocalStorage, db, "file", fullNewFilePathAbs, config.DefaultMoveFolder, "default", config, activeTorrents, queuedTorrents)
}
}
@@ -58,7 +58,7 @@ func CheckTorrentWatchFolder(c *cron.Cron, db *storm.DB, tclient *torrent.Client
}
//RefreshRSSCron refreshes all of the RSS feeds on an hourly basis
func RefreshRSSCron(c *cron.Cron, db *storm.DB, tclient *torrent.Client, torrentLocalStorage Storage.TorrentLocal, config Settings.FullClientSettings) {
func RefreshRSSCron(c *cron.Cron, db *storm.DB, tclient *torrent.Client, torrentLocalStorage Storage.TorrentLocal, config Settings.FullClientSettings, activeTorrents []string, queuedTorrents []string) {
c.AddFunc("@hourly", func() {
torrentHashHistory := Storage.FetchHashHistory(db)
RSSFeedStore := Storage.FetchRSSFeeds(db)
@@ -86,7 +86,7 @@ func RefreshRSSCron(c *cron.Cron, db *storm.DB, tclient *torrent.Client, torrent
Logger.WithFields(logrus.Fields{"err": err, "Torrent": RSSTorrent.Title}).Warn("Unable to add torrent to torrent client!")
break //break out of the loop entirely for this message since we hit an error
}
StartTorrent(clientTorrent, torrentLocalStorage, db, "magnet", "", config.DefaultMoveFolder, "RSS", config) //TODO let user specify torrent default storage location and let change on fly
AddTorrent(clientTorrent, torrentLocalStorage, db, "magnet", "", config.DefaultMoveFolder, "RSS", config, activeTorrents, queuedTorrents) //TODO let user specify torrent default storage location and let change on fly
singleFeed.Torrents = append(singleFeed.Torrents, singleRSSTorrent)
}

85
engine/engine.go
View File

@@ -129,8 +129,13 @@ func readTorrentFileFromDB(element *Storage.TorrentLocal, tclient *torrent.Clien
return singleTorrent, nil
}
//StartTorrent creates the storage.db entry and starts A NEW TORRENT and adds to the running torrent array
func StartTorrent(clientTorrent *torrent.Torrent, torrentLocalStorage Storage.TorrentLocal, torrentDbStorage *storm.DB, torrentType, torrentFilePathAbs, torrentStoragePath, labelValue string, config Settings.FullClientSettings) {
//StartTorrents attempts to start torrents by adding them to the active torrents, then the queue, and by increasing their connections (if they were stopped)
func StartTorrents(clientTorrent *torrent.Client, torrentHashes []string, activeTorrents []string, queuedTorrents []string) {
}
//AddTorrent creates the storage.db entry and starts A NEW TORRENT and adds to the running torrent array
func AddTorrent(clientTorrent *torrent.Torrent, torrentLocalStorage Storage.TorrentLocal, torrentDbStorage *storm.DB, torrentType, torrentFilePathAbs, torrentStoragePath, labelValue string, config Settings.FullClientSettings, activeTorrents []string, queuedTorrents []string) {
timedOut := timeOutInfo(clientTorrent, 45) //seeing if adding the torrent times out (giving 45 seconds)
if timedOut { //if we fail to add the torrent return
return
@@ -177,18 +182,24 @@ func StartTorrent(clientTorrent *torrent.Torrent, torrentLocalStorage Storage.To
}
torrentLocalStorage.TorrentFilePriority = TorrentFilePriorityArray
Storage.AddTorrentLocalStorage(torrentDbStorage, torrentLocalStorage) //writing all of the data to the database
clientTorrent.DownloadAll() //set all pieces to download
NumPieces := clientTorrent.NumPieces() //find the number of pieces
clientTorrent.CancelPieces(1, NumPieces) //cancel all of the pieces to use file priority
for _, singleFile := range clientTorrent.Files() { //setting all of the file priorities to normal
singleFile.SetPriority(torrent.PiecePriorityNormal)
if len(activeTorrents) < config.MaxActiveTorrents {
if len(queuedTorrents) > 0 {
queuedTorrents = append(queuedTorrents, clientTorrent.InfoHash().String())
CreateServerPushMessage(ServerPushMessage{MessageType: "serverPushMessage", MessageLevel: "success", Payload: "Torrent queued!"}, Conn)
torrentLocalStorage.QueuedStatus = "Queued"
} else {
clientTorrent.NewReader()
activeTorrents = append(activeTorrents, clientTorrent.InfoHash().String())
CreateServerPushMessage(ServerPushMessage{MessageType: "serverPushMessage", MessageLevel: "success", Payload: "Torrent added!"}, Conn)
torrentLocalStorage.QueuedStatus = "Active"
}
}
CreateServerPushMessage(ServerPushMessage{MessageType: "serverPushMessage", MessageLevel: "success", Payload: "Torrent added!"}, Conn)
Storage.AddTorrentLocalStorage(torrentDbStorage, torrentLocalStorage) //writing all of the data to the database
}
//CreateInitialTorrentArray adds all the torrents on program start from the database
func CreateInitialTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Storage.TorrentLocal, db *storm.DB) {
func CreateInitialTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Storage.TorrentLocal, db *storm.DB, config Settings.FullClientSettings, activeTorrents []string, queuedTorrents []string) {
for _, singleTorrentFromStorage := range TorrentLocalArray {
var singleTorrent *torrent.Torrent
var err error
@@ -218,20 +229,44 @@ func CreateInitialTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Sto
if err != nil {
Logger.WithFields(logrus.Fields{"torrentFile": singleTorrent.Name(), "error": err}).Error("Unable to add infobytes to the torrent!")
}
if singleTorrentFromStorage.QueuedStatus == "Active" {
singleTorrent.NewReader()
activeTorrents = append(activeTorrents, singleTorrent.InfoHash().String())
} else if singleTorrentFromStorage.QueuedStatus == "Queued" {
queuedTorrents = append(queuedTorrents, singleTorrent.InfoHash().String())
}
if singleTorrentFromStorage.TorrentStatus != "Completed" && singleTorrentFromStorage.TorrentStatus != "Stopped" {
singleTorrent.DownloadAll() //set all of the pieces to download (piece prio is NE to file prio)
NumPieces := singleTorrent.NumPieces() //find the number of pieces
singleTorrent.CancelPieces(1, NumPieces) //cancel all of the pieces to use file priority
for _, singleFile := range singleTorrent.Files() { //setting all of the file priorities to normal
singleFile.SetPriority(torrent.PiecePriorityNormal)
fmt.Println("checking about queueing torrent")
if len(activeTorrents) < config.MaxActiveTorrents {
fmt.Println("ActiveTOrrents", activeTorrents)
singleTorrent.NewReader()
singleTorrentFromStorage.QueuedStatus = "Active"
activeTorrents = append(activeTorrents, singleTorrent.InfoHash().String()) //adding the torrent hash to the queue
} else {
queuedTorrents = append(queuedTorrents, singleTorrent.InfoHash().String())
fmt.Println("Queuing torrent")
singleTorrentFromStorage.QueuedStatus = "Queued"
}
}
Storage.UpdateStorageTick(db, *singleTorrentFromStorage)
}
if len(activeTorrents) < config.MaxActiveTorrents && len(queuedTorrents) > 0 { //after all the torrents are added, see if out active torrent list isn't full, then add from the queue
fmt.Println("adding torrents from queue (if any in there)")
maxCanSend := config.MaxActiveTorrents - len(activeTorrents)
torrentsToStart := make([]string, maxCanSend)
for i, torrentHash := range queuedTorrents {
torrentsToStart[i] = torrentHash
}
StartTorrents(tclient, torrentsToStart, activeTorrents, queuedTorrents)
}
SetFilePriority(tclient, db) //Setting the desired file priority from storage
}
//CreateRunningTorrentArray creates the entire torrent list to pass to client
func CreateRunningTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Storage.TorrentLocal, PreviousTorrentArray []ClientDB, config Settings.FullClientSettings, db *storm.DB) (RunningTorrentArray []ClientDB) {
func CreateRunningTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Storage.TorrentLocal, PreviousTorrentArray []ClientDB, config Settings.FullClientSettings, db *storm.DB, activeTorrents []string, queuedTorrents []string) (RunningTorrentArray []ClientDB) {
for _, singleTorrentFromStorage := range TorrentLocalArray {
var singleTorrent *torrent.Torrent
@@ -241,11 +276,8 @@ func CreateRunningTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Sto
singleTorrent = liveTorrent
}
}
tickUpdateStruct := Storage.TorrentLocal{} //we are shoving the tick updates into a torrentlocal struct to pass to storage happens at the end of the routine
fullClientDB := new(ClientDB)
//singleTorrentStorageInfo := Storage.FetchTorrentFromStorage(db, TempHash.String()) //pulling the single torrent info from storage ()
if singleTorrentFromStorage.TorrentStatus == "Dropped" {
Logger.WithFields(logrus.Fields{"selection": singleTorrentFromStorage.TorrentName}).Info("Deleting just the torrent")
@@ -253,7 +285,7 @@ func CreateRunningTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Sto
Storage.DelTorrentLocalStorage(db, singleTorrentFromStorage.Hash)
}
if singleTorrentFromStorage.TorrentStatus == "DroppedData" {
Logger.WithFields(logrus.Fields{"selection": singleTorrentFromStorage.TorrentName}).Info("Deleting just the torrent")
Logger.WithFields(logrus.Fields{"selection": singleTorrentFromStorage.TorrentName}).Info("Deleting torrent and data")
singleTorrent.Drop()
Storage.DelTorrentLocalStorageAndFiles(db, singleTorrentFromStorage.Hash, Config.TorrentConfig.DataDir)
@@ -319,10 +351,21 @@ func CreateRunningTorrentArray(tclient *torrent.Client, TorrentLocalArray []*Sto
}
CalculateTorrentETA(singleTorrentFromStorage.TorrentSize, calculatedCompletedSize, fullClientDB) //needs to be here since we need the speed calculated before we can estimate the eta.
if (len(activeTorrents) < config.MaxActiveTorrents) && (len(queuedTorrents) > 0) { //If there is room for another torrent in active torrents, add it.
newTorrentHash := queuedTorrents[0]
fmt.Println("Moving Torrent to active queue")
for _, torrent := range tclient.Torrents() {
if newTorrentHash == torrent.InfoHash().String() {
torrent.NewReader()
activeTorrents = append(activeTorrents, newTorrentHash)
singleTorrentFromStorage.QueuedStatus = "Active"
}
}
}
fullClientDB.TotalUploadedSize = HumanizeBytes(float32(fullClientDB.TotalUploadedBytes))
fullClientDB.UploadRatio = CalculateUploadRatio(singleTorrent, fullClientDB) //calculate the upload ratio
CalculateTorrentStatus(singleTorrent, fullClientDB, config, singleTorrentFromStorage, calculatedCompletedSize, calculatedTotalSize)
CalculateTorrentStatus(singleTorrent, fullClientDB, config, singleTorrentFromStorage, calculatedCompletedSize, calculatedTotalSize, activeTorrents, queuedTorrents) //add torrents to the queue, remove from queue, etc
tickUpdateStruct.UploadRatio = fullClientDB.UploadRatio
tickUpdateStruct.TorrentSize = calculatedTotalSize

15
engine/engineHelpers.go
View File

@@ -185,17 +185,28 @@ func CalculateUploadRatio(t *torrent.Torrent, c *ClientDB) string {
}
//CalculateTorrentStatus is used to determine what the STATUS column of the frontend will display ll2
func CalculateTorrentStatus(t *torrent.Torrent, c *ClientDB, config Settings.FullClientSettings, tFromStorage *storage.TorrentLocal, bytesCompleted int64, totalSize int64) {
func CalculateTorrentStatus(t *torrent.Torrent, c *ClientDB, config Settings.FullClientSettings, tFromStorage *storage.TorrentLocal, bytesCompleted int64, totalSize int64, activeTorrents []string, queuedTorrents []string) {
if (tFromStorage.TorrentStatus == "Stopped") || (float64(c.TotalUploadedBytes)/float64(bytesCompleted) >= config.SeedRatioStop && tFromStorage.TorrentUploadLimit == true) { //If storage shows torrent stopped or if it is over the seeding ratio AND is under the global limit
c.Status = "Stopped"
c.MaxConnections = 0
t.SetMaxEstablishedConns(0)
for i, hash := range activeTorrents { //If the torrent is stopped, pull it from the active torrent array
if tFromStorage.Hash == hash {
activeTorrents = append(activeTorrents[:i], activeTorrents[i+1:]...)
}
}
} else { //Only has 2 states in storage, stopped or running, so we know it should be running, and the websocket request handled updating the database with connections and status
for _, torrentHash := range queuedTorrents {
if tFromStorage.Hash == torrentHash {
fmt.Println("Setting torrent to queued")
c.Status = "Queued"
return
}
}
bytesMissing := totalSize - bytesCompleted
c.MaxConnections = 80
t.SetMaxEstablishedConns(80)
//t.DownloadAll() //ensure that we are setting the torrent to download
if t.Seeding() && t.Stats().ActivePeers > 0 && bytesMissing == 0 {
c.Status = "Seeding"
} else if t.Stats().ActivePeers > 0 && bytesMissing > 0 {