145 lines
2.8 KiB
Go
145 lines
2.8 KiB
Go
package transfer
|
|
|
|
import (
|
|
"errors"
|
|
"io"
|
|
"sync"
|
|
"time"
|
|
|
|
"goexplore/internal/explorer"
|
|
)
|
|
|
|
type Status string
|
|
|
|
const (
|
|
StatusQueued Status = "queued"
|
|
StatusActive Status = "active"
|
|
StatusComplete Status = "complete"
|
|
StatusFailed Status = "failed"
|
|
)
|
|
|
|
type Transfer struct {
|
|
ID string `json:"id"`
|
|
Source string `json:"source"`
|
|
Destination string `json:"destination"`
|
|
Filename string `json:"filename"`
|
|
BytesTotal int64 `json:"bytes_total"`
|
|
BytesDone int64 `json:"bytes_done"`
|
|
SpeedMBps float64 `json:"speed_mbps"`
|
|
ETA int `json:"eta_seconds"`
|
|
Status Status `json:"status"`
|
|
Error string `json:"error,omitempty"`
|
|
|
|
srcExp explorer.Explorer
|
|
dstExp explorer.Explorer
|
|
|
|
startTime time.Time
|
|
}
|
|
|
|
type Manager struct {
|
|
mu sync.Mutex
|
|
transfers map[string]*Transfer
|
|
queue chan *Transfer
|
|
concurrency int
|
|
}
|
|
|
|
func NewManager(concurrency int) *Manager {
|
|
m := &Manager{
|
|
transfers: make(map[string]*Transfer),
|
|
queue: make(chan *Transfer, 100),
|
|
concurrency: concurrency,
|
|
}
|
|
for i := 0; i < concurrency; i++ {
|
|
go m.worker()
|
|
}
|
|
return m
|
|
}
|
|
|
|
func (m *Manager) worker() {
|
|
for t := range m.queue {
|
|
m.process(t)
|
|
}
|
|
}
|
|
|
|
func (m *Manager) process(t *Transfer) {
|
|
t.Status = StatusActive
|
|
t.startTime = time.Now()
|
|
err := m.doTransfer(t)
|
|
if err != nil {
|
|
t.Status = StatusFailed
|
|
t.Error = err.Error()
|
|
} else {
|
|
t.Status = StatusComplete
|
|
t.BytesDone = t.BytesTotal
|
|
}
|
|
}
|
|
|
|
func (m *Manager) doTransfer(t *Transfer) error {
|
|
r, err := t.srcExp.ReadFile(t.Source)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer r.Close()
|
|
|
|
pr := &progressReader{
|
|
r: r,
|
|
t: t,
|
|
}
|
|
|
|
return t.dstExp.WriteFile(t.Destination, pr, t.BytesTotal)
|
|
}
|
|
|
|
type progressReader struct {
|
|
r io.Reader
|
|
t *Transfer
|
|
}
|
|
|
|
func (pr *progressReader) Read(p []byte) (int, error) {
|
|
n, err := pr.r.Read(p)
|
|
if n > 0 {
|
|
pr.t.BytesDone += int64(n)
|
|
elapsed := time.Since(pr.t.startTime).Seconds()
|
|
if elapsed > 0 {
|
|
pr.t.SpeedMBps = (float64(pr.t.BytesDone) / 1024 / 1024) / elapsed
|
|
if pr.t.SpeedMBps > 0 {
|
|
pr.t.ETA = int(float64(pr.t.BytesTotal-pr.t.BytesDone) / 1024 / 1024 / pr.t.SpeedMBps)
|
|
}
|
|
}
|
|
}
|
|
return n, err
|
|
}
|
|
|
|
func (m *Manager) QueueTransfer(id, srcPath, dstPath, filename string, size int64, srcExp, dstExp explorer.Explorer) error {
|
|
if srcExp == nil || dstExp == nil {
|
|
return errors.New("invalid explorers")
|
|
}
|
|
|
|
t := &Transfer{
|
|
ID: id,
|
|
Source: srcPath,
|
|
Destination: dstPath,
|
|
Filename: filename,
|
|
BytesTotal: size,
|
|
Status: StatusQueued,
|
|
srcExp: srcExp,
|
|
dstExp: dstExp,
|
|
}
|
|
|
|
m.mu.Lock()
|
|
m.transfers[id] = t
|
|
m.mu.Unlock()
|
|
|
|
m.queue <- t
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) GetTransfers() []*Transfer {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
res := make([]*Transfer, 0, len(m.transfers))
|
|
for _, t := range m.transfers {
|
|
res = append(res, t)
|
|
}
|
|
return res
|
|
}
|