Commit 7bdb5546 authored by Michael Avila's avatar Michael Avila

Refactor per code climate rules

License: MIT
Signed-off-by: default avatarMichael Avila <davidmichaelavila@gmail.com>
parent 90d30898
...@@ -2,8 +2,9 @@ package provider ...@@ -2,8 +2,9 @@ package provider
import "github.com/ipfs/go-cid" import "github.com/ipfs/go-cid"
type offlineProvider struct {} type offlineProvider struct{}
// NewOfflineProvider creates a Provider that does nothing
func NewOfflineProvider() Provider { func NewOfflineProvider() Provider {
return &offlineProvider{} return &offlineProvider{}
} }
......
...@@ -18,13 +18,12 @@ const ( ...@@ -18,13 +18,12 @@ const (
provideOutgoingWorkerLimit = 8 provideOutgoingWorkerLimit = 8
) )
// Provider announces blocks to the network
type Provider interface { type Provider interface {
Run() Run()
Provide(cid.Cid) error Provide(cid.Cid) error
} }
// Provider announces blocks to the network, tracks which blocks are
// being provided, and untracks blocks when they're no longer in the blockstore.
type provider struct { type provider struct {
ctx context.Context ctx context.Context
// the CIDs for which provide announcements should be made // the CIDs for which provide announcements should be made
...@@ -33,6 +32,7 @@ type provider struct { ...@@ -33,6 +32,7 @@ type provider struct {
contentRouting routing.ContentRouting contentRouting routing.ContentRouting
} }
// NewProvider creates a provider that announces blocks to the network using a content router
func NewProvider(ctx context.Context, queue *Queue, contentRouting routing.ContentRouting) Provider { func NewProvider(ctx context.Context, queue *Queue, contentRouting routing.ContentRouting) Provider {
return &provider{ return &provider{
ctx: ctx, ctx: ctx,
......
...@@ -22,6 +22,7 @@ type Entry struct { ...@@ -22,6 +22,7 @@ type Entry struct {
queue *Queue queue *Queue
} }
// Complete the entry by removing it from the queue
func (e *Entry) Complete() error { func (e *Entry) Complete() error {
return e.queue.remove(e.key) return e.queue.remove(e.key)
} }
...@@ -50,9 +51,10 @@ type Queue struct { ...@@ -50,9 +51,10 @@ type Queue struct {
isRunning bool isRunning bool
} }
func NewQueue(name string, ctx context.Context, datastore ds.Datastore) (*Queue, error) { // NewQueue creates a queue for cids
namespaced := namespace.Wrap(datastore, ds.NewKey("/" + name + "/queue/")) func NewQueue(ctx context.Context, name string, datastore ds.Datastore) (*Queue, error) {
head, tail, err := getQueueHeadTail(name, ctx, namespaced) namespaced := namespace.Wrap(datastore, ds.NewKey("/"+name+"/queue/"))
head, tail, err := getQueueHeadTail(ctx, name, namespaced)
if err != nil { if err != nil {
return nil, err return nil, err
} }
...@@ -70,7 +72,7 @@ func NewQueue(name string, ctx context.Context, datastore ds.Datastore) (*Queue, ...@@ -70,7 +72,7 @@ func NewQueue(name string, ctx context.Context, datastore ds.Datastore) (*Queue,
return q, nil return q, nil
} }
// Put a cid in the queue // Enqueue puts a cid in the queue
func (q *Queue) Enqueue(cid cid.Cid) error { func (q *Queue) Enqueue(cid cid.Cid) error {
q.lock.Lock() q.lock.Lock()
defer q.lock.Unlock() defer q.lock.Unlock()
...@@ -95,21 +97,18 @@ func (q *Queue) Enqueue(cid cid.Cid) error { ...@@ -95,21 +97,18 @@ func (q *Queue) Enqueue(cid cid.Cid) error {
return nil return nil
} }
// Remove an entry from the queue. // Dequeue returns a channel that if listened to will remove entries from the queue
func (q *Queue) Dequeue() <-chan *Entry { func (q *Queue) Dequeue() <-chan *Entry {
return q.dequeue return q.dequeue
} }
// IsEmpty returns whether or not the queue has any items
func (q *Queue) IsEmpty() bool { func (q *Queue) IsEmpty() bool {
return (q.tail - q.head) == 0 return (q.tail - q.head) == 0
} }
func (q *Queue) remove(key ds.Key) error { // Run dequeues items when the dequeue channel is available to
return q.datastore.Delete(key) // be written to.
}
// dequeue items when the dequeue channel is available to
// be written to
func (q *Queue) Run() { func (q *Queue) Run() {
q.isRunning = true q.isRunning = true
go func() { go func() {
...@@ -178,7 +177,7 @@ func (q *Queue) next() (*Entry, error) { ...@@ -178,7 +177,7 @@ func (q *Queue) next() (*Entry, error) {
return nil, err return nil, err
} }
entry := &Entry { entry := &Entry{
cid: id, cid: id,
key: nextKey, key: nextKey,
queue: q, queue: q,
...@@ -194,14 +193,14 @@ func (q *Queue) queueKey(id uint64) ds.Key { ...@@ -194,14 +193,14 @@ func (q *Queue) queueKey(id uint64) ds.Key {
} }
// crawl over the queue entries to find the head and tail // crawl over the queue entries to find the head and tail
func getQueueHeadTail(name string, ctx context.Context, datastore ds.Datastore) (uint64, uint64, error) { func getQueueHeadTail(ctx context.Context, name string, datastore ds.Datastore) (uint64, uint64, error) {
query := query.Query{} query := query.Query{}
results, err := datastore.Query(query) results, err := datastore.Query(query)
if err != nil { if err != nil {
return 0, 0, err return 0, 0, err
} }
var tail uint64 = 0 var tail uint64
var head uint64 = math.MaxUint64 var head uint64 = math.MaxUint64
for entry := range results.Next() { for entry := range results.Next() {
select { select {
...@@ -219,8 +218,8 @@ func getQueueHeadTail(name string, ctx context.Context, datastore ds.Datastore) ...@@ -219,8 +218,8 @@ func getQueueHeadTail(name string, ctx context.Context, datastore ds.Datastore)
head = id head = id
} }
if (id+1) > tail { if (id + 1) > tail {
tail = (id+1) tail = (id + 1)
} }
} }
if err := results.Close(); err != nil { if err := results.Close(); err != nil {
...@@ -233,3 +232,6 @@ func getQueueHeadTail(name string, ctx context.Context, datastore ds.Datastore) ...@@ -233,3 +232,6 @@ func getQueueHeadTail(name string, ctx context.Context, datastore ds.Datastore)
return head, tail, nil return head, tail, nil
} }
func (q *Queue) remove(key ds.Key) error {
return q.datastore.Delete(key)
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment