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

make MemoryStore consistent with the New...() pattern

Browse source
This commit is contained in:
Alex Grintsvayg 2019-10-03 16:47:23 -04:00
parent 2ca83139df
commit 69f1e0f4ca
No known key found for this signature in database
GPG key ID: AEB3F089F86A22B5
8 changed files with 34 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/test.go
View file

@ -29,7 +29,7 @@ func init() {
func testCmd(cmd *cobra.Command, args []string) {
log.Printf("reflector version %s", meta.Version)
memStore := &store.MemoryBlobStore{}
memStore := store.NewMemoryBlobStore()
reflectorServer := reflector.NewServer(memStore)
reflectorServer.Timeout = 3 * time.Minute

23
peer/server.go
View file

@ -8,13 +8,12 @@ import (
"strings"
"time"
"github.com/lbryio/lbry.go/stream"
"github.com/lbryio/reflector.go/reflector"
"github.com/lbryio/reflector.go/store"
"github.com/lbryio/lbry.go/extras/errors"
"github.com/lbryio/lbry.go/extras/stop"
"github.com/lbryio/lbry.go/stream"
log "github.com/sirupsen/logrus"
)
@ -169,14 +168,14 @@ func (s *Server) handleAvailabilityRequest(data []byte) ([]byte, error) {
var request availabilityRequest
err := json.Unmarshal(data, &request)
if err != nil {
return []byte{}, err
return nil, err
}
availableBlobs := []string{}
for _, blobHash := range request.RequestedBlobs {
exists, err := s.store.Has(blobHash)
if err != nil {
return []byte{}, err
return nil, err
}
if exists {
availableBlobs = append(availableBlobs, blobHash)
@ -190,7 +189,7 @@ func (s *Server) handleAvailabilityRequest(data []byte) ([]byte, error) {
// var request paymentRateRequest
// err := json.Unmarshal(data, &request)
// if err != nil {
// return []byte{}, err
// return nil, err
// }
//
// offerReply := paymentRateAccepted
@ -205,14 +204,14 @@ func (s *Server) handleAvailabilityRequest(data []byte) ([]byte, error) {
// var request blobRequest
// err := json.Unmarshal(data, &request)
// if err != nil {
// return []byte{}, err
// return nil, err
// }
//
// log.Debugln("Sending blob " + request.RequestedBlob[:8])
//
// blob, err := s.store.Get(request.RequestedBlob)
// if err != nil {
// return []byte{}, err
// return nil, err
// }
//
// response, err := json.Marshal(blobResponse{IncomingBlob: incomingBlob{
@ -220,7 +219,7 @@ func (s *Server) handleAvailabilityRequest(data []byte) ([]byte, error) {
// Length: len(blob),
// }})
// if err != nil {
// return []byte{}, err
// return nil, err
// }
//
// return append(response, blob...), nil
@ -230,7 +229,7 @@ func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
var request compositeRequest
err := json.Unmarshal(data, &request)
if err != nil {
return []byte{}, err
return nil, err
}
response := compositeResponse{
@ -242,7 +241,7 @@ func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
for _, blobHash := range request.RequestedBlobs {
exists, err := s.store.Has(blobHash)
if err != nil {
return []byte{}, err
return nil, err
}
if exists {
availableBlobs = append(availableBlobs, blobHash)
@ -270,7 +269,7 @@ func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
Error: err.Error(),
}
} else if err != nil {
return []byte{}, err
return nil, err
} else {
response.IncomingBlob = incomingBlob{
BlobHash: reflector.BlobHash(blob),
@ -282,7 +281,7 @@ func (s *Server) handleCompositeRequest(data []byte) ([]byte, error) {
respData, err := json.Marshal(response)
if err != nil {
return []byte{}, err
return nil, err
}
return append(respData, blob...), nil

4
peer/server_test.go
View file

@ -34,7 +34,7 @@ var availabilityRequests = []pair{
}
func getServer(t *testing.T, withBlobs bool) *Server {
st := store.MemoryBlobStore{}
st := store.NewMemoryBlobStore()
if withBlobs {
for k, v := range blobs {
err := st.Put(k, v)
@ -43,7 +43,7 @@ func getServer(t *testing.T, withBlobs bool) *Server {
}
}
}
return NewServer(&st)
return NewServer(st)
}
func TestAvailabilityRequest_NoBlobs(t *testing.T) {

8
reflector/server_test.go
View file

@ -22,7 +22,7 @@ func startServerOnRandomPort(t *testing.T) (*Server, int) {
t.Fatal(err)
}
srv := NewServer(&store.MemoryBlobStore{})
srv := NewServer(store.NewMemoryBlobStore())
err = srv.Start("127.0.0.1:" + strconv.Itoa(port))
if err != nil {
t.Fatal(err)
@ -119,7 +119,7 @@ func TestServer_Timeout(t *testing.T) {
t.Fatal(err)
}
srv := NewServer(&store.MemoryBlobStore{})
srv := NewServer(store.NewMemoryBlobStore())
srv.Timeout = testTimeout
err = srv.Start("127.0.0.1:" + strconv.Itoa(port))
if err != nil {
@ -161,7 +161,7 @@ func TestServer_Timeout(t *testing.T) {
//}
type mockPartialStore struct {
store.MemoryBlobStore
*store.MemoryBlobStore
missing []string
}
@ -181,7 +181,7 @@ func TestServer_PartialUpload(t *testing.T) {
missing[i] = bits.Rand().String()
}
st := store.BlobStore(&mockPartialStore{missing: missing})
st := store.BlobStore(&mockPartialStore{MemoryBlobStore: store.NewMemoryBlobStore(), missing: missing})
if _, ok := st.(neededBlobChecker); !ok {
t.Fatal("mock does not implement the relevant interface")
}

8
store/caching_test.go
View file

@ -6,8 +6,8 @@ import (
)
func TestCachingBlobStore_Put(t *testing.T) {
origin := &MemoryBlobStore{}
cache := &MemoryBlobStore{}
origin := NewMemoryBlobStore()
cache := NewMemoryBlobStore()
s := NewCachingBlobStore(origin, cache)
b := []byte("this is a blob of stuff")
@ -36,8 +36,8 @@ func TestCachingBlobStore_Put(t *testing.T) {
}
func TestCachingBlobStore_CacheMiss(t *testing.T) {
origin := &MemoryBlobStore{}
cache := &MemoryBlobStore{}
origin := NewMemoryBlobStore()
cache := NewMemoryBlobStore()
s := NewCachingBlobStore(origin, cache)
b := []byte("this is a blob of stuff")

21
store/memory.go
View file

@ -7,35 +7,32 @@ import (
// MemoryBlobStore is an in memory only blob store with no persistence.
type MemoryBlobStore struct {
blobs map[string][]byte
blobs map[string]stream.Blob
}
func NewMemoryBlobStore() *MemoryBlobStore {
return &MemoryBlobStore{
blobs: make(map[string]stream.Blob),
}
}
// Has returns T/F if the blob is currently stored. It will never error.
func (m *MemoryBlobStore) Has(hash string) (bool, error) {
if m.blobs == nil {
m.blobs = make(map[string][]byte)
}
_, ok := m.blobs[hash]
return ok, nil
}
// Get returns the blob byte slice if present and errors if the blob is not found.
func (m *MemoryBlobStore) Get(hash string) (stream.Blob, error) {
if m.blobs == nil {
m.blobs = make(map[string][]byte)
}
blob, ok := m.blobs[hash]
if !ok {
return []byte{}, errors.Err(ErrBlobNotFound)
return nil, errors.Err(ErrBlobNotFound)
}
return blob, nil
}
// Put stores the blob in memory
func (m *MemoryBlobStore) Put(hash string, blob stream.Blob) error {
if m.blobs == nil {
m.blobs = make(map[string][]byte)
}
m.blobs[hash] = blob
return nil
}
@ -52,6 +49,6 @@ func (m *MemoryBlobStore) Delete(hash string) error {
}
// Debug returns the blobs in memory. It's useful for testing and debugging.
func (m *MemoryBlobStore) Debug() map[string][]byte {
func (m *MemoryBlobStore) Debug() map[string]stream.Blob {
return m.blobs
}

4
store/memory_test.go
View file

@ -8,7 +8,7 @@ import (
)
func TestMemoryBlobStore_Put(t *testing.T) {
s := MemoryBlobStore{}
s := NewMemoryBlobStore()
blob := []byte("abcdefg")
err := s.Put("abc", blob)
if err != nil {
@ -17,7 +17,7 @@ func TestMemoryBlobStore_Put(t *testing.T) {
}
func TestMemoryBlobStore_Get(t *testing.T) {
s := MemoryBlobStore{}
s := NewMemoryBlobStore()
hash := "abc"
blob := []byte("abcdefg")
err := s.Put(hash, blob)

2
store/store.go
View file

@ -5,7 +5,7 @@ import (
"github.com/lbryio/lbry.go/stream"
)
// BlobStore is an interface with methods for consistently handling blob storage.
// BlobStore is an interface for handling blob storage.
type BlobStore interface {
// Does blob exist in the store
Has(hash string) (bool, error)