LBRYCommunity/reflector.go
1
0
Fork 0
Code Issues 7 Pull requests Projects Releases 10 Packages Wiki Activity

make db-backed store more generic (not specific to s3)

Browse source
This commit is contained in:
Alex Grintsvayg 2019-10-03 16:58:17 -04:00
parent 69f1e0f4ca
commit 661c20a21d
No known key found for this signature in database
GPG key ID: AEB3F089F86A22B5
6 changed files with 28 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/peer.go
View file

@ -37,7 +37,7 @@ func peerCmd(cmd *cobra.Command, args []string) {
err = db.Connect(globalConfig.DBConn) err = db.Connect(globalConfig.DBConn)
checkErr(err) checkErr(err)
combo := store.NewDBBackedS3Store(s3, db) combo := store.NewDBBackedStore(s3, db)
peerServer = peer.NewServer(combo) peerServer = peer.NewServer(combo)
} }

2
cmd/reflector.go
View file

@ -48,7 +48,7 @@ func reflectorCmd(cmd *cobra.Command, args []string) {
log.Fatal(err) log.Fatal(err)
} }
blobStore = store.NewDBBackedS3Store(s3, db) blobStore = store.NewDBBackedStore(s3, db)
reflectorServer = reflector.NewServer(blobStore) reflectorServer = reflector.NewServer(blobStore)
reflectorServer.Timeout = 3 * time.Minute reflectorServer.Timeout = 3 * time.Minute

2
cmd/start.go
View file

@ -56,7 +56,7 @@ func startCmd(cmd *cobra.Command, args []string) {
err := db.Connect(globalConfig.DBConn) err := db.Connect(globalConfig.DBConn)
checkErr(err) checkErr(err)
s3 := store.NewS3BlobStore(globalConfig.AwsID, globalConfig.AwsSecret, globalConfig.BucketRegion, globalConfig.BucketName) s3 := store.NewS3BlobStore(globalConfig.AwsID, globalConfig.AwsSecret, globalConfig.BucketRegion, globalConfig.BucketName)
comboStore := store.NewDBBackedS3Store(s3, db) comboStore := store.NewDBBackedStore(s3, db)
conf := prism.DefaultConf() conf := prism.DefaultConf()

2
cmd/upload.go
View file

@ -32,7 +32,7 @@ func uploadCmd(cmd *cobra.Command, args []string) {
err := db.Connect(globalConfig.DBConn) err := db.Connect(globalConfig.DBConn)
checkErr(err) checkErr(err)
st := store.NewDBBackedS3Store( st := store.NewDBBackedStore(
store.NewS3BlobStore(globalConfig.AwsID, globalConfig.AwsSecret, globalConfig.BucketRegion, globalConfig.BucketName), store.NewS3BlobStore(globalConfig.AwsID, globalConfig.AwsSecret, globalConfig.BucketRegion, globalConfig.BucketName),
db) db)

4
reflector/uploader.go
View file

@ -30,7 +30,7 @@ type Summary struct {
type Uploader struct { type Uploader struct {
db *db.SQL db *db.SQL
store *store.DBBackedS3Store // could just be store.BlobStore interface store *store.DBBackedStore // could just be store.BlobStore interface
workers int workers int
skipExistsCheck bool skipExistsCheck bool
stopper *stop.Group stopper *stop.Group
@ -39,7 +39,7 @@ type Uploader struct {
count Summary count Summary
} }
func NewUploader(db *db.SQL, store *store.DBBackedS3Store, workers int, skipExistsCheck bool) *Uploader { func NewUploader(db *db.SQL, store *store.DBBackedStore, workers int, skipExistsCheck bool) *Uploader {
return &Uploader{ return &Uploader{
db: db, db: db,
store: store, store: store,

44
store/dbbacked.go
View file

@ -12,32 +12,32 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
// DBBackedS3Store is an instance of an S3 Store that is backed by a DB for what is stored. // DBBackedStore is a store that's backed by a DB. The DB contains data about what's in the store.
type DBBackedS3Store struct { type DBBackedStore struct {
s3 *S3BlobStore blobs BlobStore
db *db.SQL db *db.SQL
blockedMu sync.RWMutex blockedMu sync.RWMutex
blocked map[string]bool blocked map[string]bool
} }
// NewDBBackedS3Store returns an initialized store pointer. // NewDBBackedStore returns an initialized store pointer.
func NewDBBackedS3Store(s3 *S3BlobStore, db *db.SQL) *DBBackedS3Store { func NewDBBackedStore(blobs BlobStore, db *db.SQL) *DBBackedStore {
return &DBBackedS3Store{s3: s3, db: db} return &DBBackedStore{blobs: blobs, db: db}
} }
// Has returns true if the blob is in the store // Has returns true if the blob is in the store
func (d *DBBackedS3Store) Has(hash string) (bool, error) { func (d *DBBackedStore) Has(hash string) (bool, error) {
return d.db.HasBlob(hash) return d.db.HasBlob(hash)
} }
// Get gets the blob // Get gets the blob
func (d *DBBackedS3Store) Get(hash string) (stream.Blob, error) { func (d *DBBackedStore) Get(hash string) (stream.Blob, error) {
return d.s3.Get(hash) return d.blobs.Get(hash)
} }
// Put stores the blob in the S3 store and stores the blob information in the DB. // Put stores the blob in the S3 store and stores the blob information in the DB.
func (d *DBBackedS3Store) Put(hash string, blob stream.Blob) error { func (d *DBBackedStore) Put(hash string, blob stream.Blob) error {
err := d.s3.Put(hash, blob) err := d.blobs.Put(hash, blob)
if err != nil { if err != nil {
return err return err
} }
@ -47,7 +47,7 @@ func (d *DBBackedS3Store) Put(hash string, blob stream.Blob) error {
// PutSD stores the SDBlob in the S3 store. It will return an error if the sd blob is missing the stream hash or if // PutSD stores the SDBlob in the S3 store. It will return an error if the sd blob is missing the stream hash or if
// there is an error storing the blob information in the DB. // there is an error storing the blob information in the DB.
func (d *DBBackedS3Store) PutSD(hash string, blob stream.Blob) error { func (d *DBBackedStore) PutSD(hash string, blob stream.Blob) error {
var blobContents db.SdBlob var blobContents db.SdBlob
err := json.Unmarshal(blob, &blobContents) err := json.Unmarshal(blob, &blobContents)
if err != nil { if err != nil {
@ -57,7 +57,7 @@ func (d *DBBackedS3Store) PutSD(hash string, blob stream.Blob) error {
return errors.Err("sd blob is missing stream hash") return errors.Err("sd blob is missing stream hash")
} }
err = d.s3.PutSD(hash, blob) err = d.blobs.PutSD(hash, blob)
if err != nil { if err != nil {
return err return err
} }
@ -65,8 +65,8 @@ func (d *DBBackedS3Store) PutSD(hash string, blob stream.Blob) error {
return d.db.AddSDBlob(hash, len(blob), blobContents) return d.db.AddSDBlob(hash, len(blob), blobContents)
} }
func (d *DBBackedS3Store) Delete(hash string) error { func (d *DBBackedStore) Delete(hash string) error {
err := d.s3.Delete(hash) err := d.blobs.Delete(hash)
if err != nil { if err != nil {
return err return err
} }
@ -75,7 +75,7 @@ func (d *DBBackedS3Store) Delete(hash string) error {
} }
// Block deletes the blob and prevents it from being uploaded in the future // Block deletes the blob and prevents it from being uploaded in the future
func (d *DBBackedS3Store) Block(hash string) error { func (d *DBBackedStore) Block(hash string) error {
if blocked, err := d.isBlocked(hash); blocked || err != nil { if blocked, err := d.isBlocked(hash); blocked || err != nil {
return err return err
} }
@ -93,7 +93,7 @@ func (d *DBBackedS3Store) Block(hash string) error {
} }
if has { if has {
err = d.s3.Delete(hash) err = d.blobs.Delete(hash)
if err != nil { if err != nil {
return err return err
} }
@ -108,7 +108,7 @@ func (d *DBBackedS3Store) Block(hash string) error {
} }
// Wants returns false if the hash exists or is blocked, true otherwise // Wants returns false if the hash exists or is blocked, true otherwise
func (d *DBBackedS3Store) Wants(hash string) (bool, error) { func (d *DBBackedStore) Wants(hash string) (bool, error) {
blocked, err := d.isBlocked(hash) blocked, err := d.isBlocked(hash)
if blocked || err != nil { if blocked || err != nil {
return false, err return false, err
@ -121,11 +121,11 @@ func (d *DBBackedS3Store) Wants(hash string) (bool, error) {
// MissingBlobsForKnownStream returns missing blobs for an existing stream // MissingBlobsForKnownStream returns missing blobs for an existing stream
// WARNING: if the stream does NOT exist, no blob hashes will be returned, which looks // WARNING: if the stream does NOT exist, no blob hashes will be returned, which looks
// like no blobs are missing // like no blobs are missing
func (d *DBBackedS3Store) MissingBlobsForKnownStream(sdHash string) ([]string, error) { func (d *DBBackedStore) MissingBlobsForKnownStream(sdHash string) ([]string, error) {
return d.db.MissingBlobsForKnownStream(sdHash) return d.db.MissingBlobsForKnownStream(sdHash)
} }
func (d *DBBackedS3Store) markBlocked(hash string) error { func (d *DBBackedStore) markBlocked(hash string) error {
err := d.initBlocked() err := d.initBlocked()
if err != nil { if err != nil {
return err return err
@ -138,7 +138,7 @@ func (d *DBBackedS3Store) markBlocked(hash string) error {
return nil return nil
} }
func (d *DBBackedS3Store) isBlocked(hash string) (bool, error) { func (d *DBBackedStore) isBlocked(hash string) (bool, error) {
err := d.initBlocked() err := d.initBlocked()
if err != nil { if err != nil {
return false, err return false, err
@ -150,7 +150,7 @@ func (d *DBBackedS3Store) isBlocked(hash string) (bool, error) {
return d.blocked[hash], nil return d.blocked[hash], nil
} }
func (d *DBBackedS3Store) initBlocked() error { func (d *DBBackedStore) initBlocked() error {
// first check without blocking since this is the most likely scenario // first check without blocking since this is the most likely scenario
if d.blocked != nil { if d.blocked != nil {
return nil return nil