provider: refactor to only maintain one batched implementation and ad…

…d throughput callback

provider/queue/queue.go → provider/internal/queue/queue.go

@@ -3,6 +3,7 @@ package queue
 import (
 	"context"
 	"fmt"
+
 	cid "github.com/ipfs/go-cid"
 	datastore "github.com/ipfs/go-datastore"
 	namespace "github.com/ipfs/go-datastore/namespace"
@@ -20,7 +21,6 @@ var log = logging.Logger("provider.queue")
 type Queue struct {
 	// used to differentiate queues in datastore
 	// e.g. provider vs reprovider
-	name    string
 	ctx     context.Context
 	ds      datastore.Datastore // Must be threadsafe
 	dequeue chan cid.Cid
@@ -32,11 +32,10 @@ type Queue struct {
 }
 
 // NewQueue creates a queue for cids
-func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue, error) {
-	namespaced := namespace.Wrap(ds, datastore.NewKey("/"+name+"/queue/"))
-	cancelCtx, cancel := context.WithCancel(ctx)
+func NewQueue(ds datastore.Datastore) *Queue {
+	namespaced := namespace.Wrap(ds, datastore.NewKey("/queue"))
+	cancelCtx, cancel := context.WithCancel(context.Background())
 	q := &Queue{
-		name:    name,
 		ctx:     cancelCtx,
 		ds:      namespaced,
 		dequeue: make(chan cid.Cid),
@@ -45,7 +44,7 @@ func NewQueue(ctx context.Context, name string, ds datastore.Datastore) (*Queue,
 		closed:  make(chan struct{}, 1),
 	}
 	q.work()
-	return q, nil
+	return q
 }
 
 // Close stops the queue

provider/queue/queue_test.go → provider/internal/queue/queue_test.go

@@ -43,10 +43,8 @@ func TestBasicOperation(t *testing.T) {
 	defer ctx.Done()
 
 	ds := sync.MutexWrap(datastore.NewMapDatastore())
-	queue, err := NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue := NewQueue(ds)
+	defer queue.Close()
 
 	cids := makeCids(10)
 
@@ -63,10 +61,8 @@ func TestMangledData(t *testing.T) {
 	defer ctx.Done()
 
 	ds := sync.MutexWrap(datastore.NewMapDatastore())
-	queue, err := NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue := NewQueue(ds)
+	defer queue.Close()
 
 	cids := makeCids(10)
 	for _, c := range cids {
@@ -75,7 +71,7 @@ func TestMangledData(t *testing.T) {
 
 	// put bad data in the queue
 	queueKey := datastore.NewKey("/test/0")
-	err = queue.ds.Put(ctx, queueKey, []byte("borked"))
+	err := queue.ds.Put(ctx, queueKey, []byte("borked"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -91,10 +87,8 @@ func TestInitialization(t *testing.T) {
 	defer ctx.Done()
 
 	ds := sync.MutexWrap(datastore.NewMapDatastore())
-	queue, err := NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue := NewQueue(ds)
+	defer queue.Close()
 
 	cids := makeCids(10)
 	for _, c := range cids {
@@ -104,10 +98,8 @@ func TestInitialization(t *testing.T) {
 	assertOrdered(cids[:5], queue, t)
 
 	// make a new queue, same data
-	queue, err = NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue = NewQueue(ds)
+	defer queue.Close()
 
 	assertOrdered(cids[5:], queue, t)
 }
@@ -118,21 +110,18 @@ func TestInitializationWithManyCids(t *testing.T) {
 	defer ctx.Done()
 
 	ds := sync.MutexWrap(datastore.NewMapDatastore())
-	queue, err := NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue := NewQueue(ds)
 
 	cids := makeCids(25)
 	for _, c := range cids {
 		queue.Enqueue(c)
 	}
 
+	queue.Close()
+
 	// make a new queue, same data
-	queue, err = NewQueue(ctx, "test", ds)
-	if err != nil {
-		t.Fatal(err)
-	}
+	queue = NewQueue(ds)
+	defer queue.Close()
 
 	assertOrdered(cids, queue, t)
 }