syncthing/lib/model/sentdownloadstate.go
2017-02-09 08:04:16 +01:00

189 lines
6.0 KiB
Go

// Copyright (C) 2015 The Syncthing Authors.
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at https://mozilla.org/MPL/2.0/.
package model
import (
"time"
"github.com/syncthing/syncthing/lib/protocol"
)
// sentFolderFileDownloadState represents a state of what we've announced as available
// to some remote device for a specific file.
type sentFolderFileDownloadState struct {
blockIndexes []int32
version protocol.Vector
updated time.Time
created time.Time
}
// sentFolderDownloadState represents a state of what we've announced as available
// to some remote device for a specific folder.
type sentFolderDownloadState struct {
files map[string]*sentFolderFileDownloadState
}
// update takes a set of currently active sharedPullerStates, and returns a list
// of updates which we need to send to the client to become up to date.
func (s *sentFolderDownloadState) update(pullers []*sharedPullerState) []protocol.FileDownloadProgressUpdate {
var name string
var updates []protocol.FileDownloadProgressUpdate
seen := make(map[string]struct{}, len(pullers))
for _, puller := range pullers {
name = puller.file.Name
seen[name] = struct{}{}
pullerBlockIndexes := puller.Available()
pullerVersion := puller.file.Version
pullerBlockIndexesUpdated := puller.AvailableUpdated()
pullerCreated := puller.created
localFile, ok := s.files[name]
// New file we haven't seen before
if !ok {
// Only send an update if the file actually has some blocks.
if len(pullerBlockIndexes) > 0 {
s.files[name] = &sentFolderFileDownloadState{
blockIndexes: pullerBlockIndexes,
updated: pullerBlockIndexesUpdated,
version: pullerVersion,
created: pullerCreated,
}
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: pullerVersion,
UpdateType: protocol.UpdateTypeAppend,
BlockIndexes: pullerBlockIndexes,
})
}
continue
}
// Existing file we've already sent an update for.
if pullerBlockIndexesUpdated.Equal(localFile.updated) && pullerVersion.Equal(localFile.version) {
// The file state hasn't changed, go to next.
continue
}
if !pullerVersion.Equal(localFile.version) || !pullerCreated.Equal(localFile.created) {
// The version has changed or the puller was reconstrcuted due to failure.
// Clean up whatever we had for the old file, and advertise the new file.
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: localFile.version,
UpdateType: protocol.UpdateTypeForget,
})
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: pullerVersion,
UpdateType: protocol.UpdateTypeAppend,
BlockIndexes: pullerBlockIndexes,
})
localFile.blockIndexes = pullerBlockIndexes
localFile.updated = pullerBlockIndexesUpdated
localFile.version = pullerVersion
localFile.created = pullerCreated
continue
}
// Relies on the fact that sharedPullerState.Available() should always
// append.
newBlocks := pullerBlockIndexes[len(localFile.blockIndexes):]
localFile.blockIndexes = append(localFile.blockIndexes, newBlocks...)
localFile.updated = pullerBlockIndexesUpdated
// If there are new blocks, send the update.
if len(newBlocks) > 0 {
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: localFile.version,
UpdateType: protocol.UpdateTypeAppend,
BlockIndexes: newBlocks,
})
}
}
// For each file that we are tracking, see if there still is a puller for it
// if not, the file completed or errored out.
for name, info := range s.files {
_, ok := seen[name]
if !ok {
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: info.version,
UpdateType: protocol.UpdateTypeForget,
})
delete(s.files, name)
}
}
return updates
}
// destroy removes all stored state, and returns a set of updates we need to
// dispatch to clean up the state on the remote end.
func (s *sentFolderDownloadState) destroy() []protocol.FileDownloadProgressUpdate {
updates := make([]protocol.FileDownloadProgressUpdate, 0, len(s.files))
for name, info := range s.files {
updates = append(updates, protocol.FileDownloadProgressUpdate{
Name: name,
Version: info.version,
UpdateType: protocol.UpdateTypeForget,
})
delete(s.files, name)
}
return updates
}
// sentDownloadState represents a state of what we've announced as available
// to some remote device. It is used from within the progress emitter
// which only has one routine, hence is deemed threadsafe.
type sentDownloadState struct {
folderStates map[string]*sentFolderDownloadState
}
// update receives a folder, and a slice of pullers that are currently available
// for the given folder, and according to the state of what we've seen before
// returns a set of updates which we should send to the remote device to make
// it aware of everything that we currently have available.
func (s *sentDownloadState) update(folder string, pullers []*sharedPullerState) []protocol.FileDownloadProgressUpdate {
fs, ok := s.folderStates[folder]
if !ok {
fs = &sentFolderDownloadState{
files: make(map[string]*sentFolderFileDownloadState),
}
s.folderStates[folder] = fs
}
return fs.update(pullers)
}
// folders returns a set of folders this state is currently aware off.
func (s *sentDownloadState) folders() []string {
folders := make([]string, 0, len(s.folderStates))
for key := range s.folderStates {
folders = append(folders, key)
}
return folders
}
// cleanup cleans up all state related to a folder, and returns a set of updates
// which would clean up the state on the remote device.
func (s *sentDownloadState) cleanup(folder string) []protocol.FileDownloadProgressUpdate {
fs, ok := s.folderStates[folder]
if ok {
updates := fs.destroy()
delete(s.folderStates, folder)
return updates
}
return nil
}