145 lines
3.5 KiB
Go
145 lines
3.5 KiB
Go
package forgepb
|
|
|
|
import (
|
|
"github.com/destel/rill"
|
|
"go.wit.com/lib/protobuf/gitpb"
|
|
"go.wit.com/log"
|
|
)
|
|
|
|
// rill is awesome. long live rill
|
|
// attempt scan with rill
|
|
func (f *Forge) rillUpdate(pool1 int, pool2 int) (int, error) {
|
|
var repos []*gitpb.Repo
|
|
all := f.Repos.SortByFullPath()
|
|
for all.Scan() {
|
|
repo := all.Next()
|
|
repos = append(repos, repo)
|
|
}
|
|
|
|
// Convert a slice of user IDs into a channel
|
|
ids := rill.FromSlice(repos, nil)
|
|
|
|
// Read users from the API.
|
|
// Concurrency = 20
|
|
rills := rill.Map(ids, pool1, func(repo *gitpb.Repo) (*gitpb.Repo, error) {
|
|
return repo, nil
|
|
})
|
|
|
|
var counter int
|
|
// Activate users.
|
|
// Concurrency = 10
|
|
err := rill.ForEach(rills, pool2, func(repo *gitpb.Repo) error {
|
|
counter += 1
|
|
// log.Info("rill.ForEach() gopath=", repo.GetGoPath())
|
|
return f.updateRepo(repo)
|
|
})
|
|
|
|
return counter, err
|
|
}
|
|
|
|
func (f *Forge) updateRepo(repo *gitpb.Repo) error {
|
|
if !repo.IsValidDir() {
|
|
log.Printf("%10s %-50s gopath=%s\n", "git dir is missing:", repo.FullPath, repo.GetGoPath())
|
|
f.Repos.DeleteByFullPath(repo.FullPath)
|
|
f.configSave = true
|
|
return nil
|
|
}
|
|
|
|
if repo.DidRepoChange() {
|
|
f.configSave = true
|
|
log.Info("repo changed ", repo.FullPath, repo.StateChange)
|
|
if err := repo.Reload(); err != nil {
|
|
return err
|
|
}
|
|
} else {
|
|
// log.Info("repo did not change", repo.FullPath, repo.StateChange)
|
|
}
|
|
if f.Config.IsReadOnly(repo.GetGoPath()) {
|
|
if repo.ReadOnly {
|
|
} else {
|
|
log.Info("readonly flag on repo is wrong", repo.GetGoPath())
|
|
repo.ReadOnly = true
|
|
f.configSave = true
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var RillX int = 30
|
|
var RillY int = 10
|
|
|
|
// x is the size of the queued up pool (shouldn't matter here for this I think)
|
|
// y is how many simultanous functions will run
|
|
// todo: tune and compute x,y by # of CPUs and disk io
|
|
// todo: store x,y in forge config ? (or compute them. notsure)
|
|
func (f *Forge) RillReload() int {
|
|
var all []*gitpb.Repo
|
|
tmp := f.Repos.All()
|
|
for tmp.Scan() {
|
|
repo := tmp.Next()
|
|
if !repo.IsValidDir() {
|
|
log.Printf("%s %-50s", "got an invalid repo in forgepb.RillFuncError()", repo.GetGoPath())
|
|
continue
|
|
}
|
|
all = append(all, repo)
|
|
}
|
|
// Convert a slice of user IDs into a channel
|
|
ids := rill.FromSlice(all, nil)
|
|
|
|
var counter int
|
|
// Read users from the API.
|
|
// Concurrency = 20
|
|
dirs := rill.Map(ids, RillX, func(repo *gitpb.Repo) (*gitpb.Repo, error) {
|
|
return repo, nil
|
|
})
|
|
|
|
rill.ForEach(dirs, RillY, func(repo *gitpb.Repo) error {
|
|
if !repo.DidRepoChange() {
|
|
return nil
|
|
}
|
|
f.configSave = true
|
|
repo.Reload()
|
|
counter += 1
|
|
return nil
|
|
})
|
|
|
|
return counter
|
|
}
|
|
|
|
// x is the size of the queued up pool (shouldn't matter here for this I think)
|
|
// y is how many simultanous functions will run
|
|
// todo: tune and compute x,y by # of CPUs and disk io
|
|
// todo: store x,y in forge config ? (or compute them. notsure)
|
|
func (f *Forge) RillFuncError(rillf func(*gitpb.Repo) error) int {
|
|
var all []*gitpb.Repo
|
|
tmp := f.Repos.All()
|
|
for tmp.Scan() {
|
|
repo := tmp.Next()
|
|
if !repo.IsValidDir() {
|
|
log.Printf("%s %-50s", "got an invalid repo in forgepb.RillFuncError()", repo.GetGoPath())
|
|
continue
|
|
}
|
|
all = append(all, repo)
|
|
}
|
|
// Convert a slice of user IDs into a channel
|
|
ids := rill.FromSlice(all, nil)
|
|
|
|
var counter int
|
|
// Read users from the API.
|
|
// Concurrency = 20
|
|
dirs := rill.Map(ids, RillX, func(id *gitpb.Repo) (*gitpb.Repo, error) {
|
|
return id, nil
|
|
})
|
|
|
|
err := rill.ForEach(dirs, RillY, func(repo *gitpb.Repo) error {
|
|
counter += 1
|
|
return rillf(repo)
|
|
})
|
|
|
|
if err != nil {
|
|
log.Info("rill.ForEach() error:", err)
|
|
}
|
|
|
|
return counter
|
|
}
|