simplify pointer passing for interface in cursor

Author: umputun

pool/cursor.go

@@ -1,6 +1,10 @@
 package pool
 
-import "context"
+import (
+	"context"
+	"errors"
+	"reflect"
+)
 
 // Cursor provides synchronous access to async data from pool's response channel
 type Cursor struct {
@@ -10,7 +14,7 @@ type Cursor struct {
 
 // Next returns next result from the cursor, ok = false on completion.
 // Any error saved internally and can be returned by Err call
-func (c *Cursor) Next(ctx context.Context, v *interface{}) bool {
+func (c *Cursor) Next(ctx context.Context, v interface{}) bool {
 	for {
 		select {
 		case resp, ok := <-c.ch:
@@ -21,7 +25,14 @@ func (c *Cursor) Next(ctx context.Context, v *interface{}) bool {
 				c.err = resp.err
 				continue
 			}
-			*v = resp.value
+
+			rv := reflect.ValueOf(v)
+			if rv.Kind() != reflect.Ptr || rv.IsNil() {
+				c.err = errors.New("value type is not pointer")
+				return false
+			}
+			dstValue := reflect.Indirect(rv)
+			dstValue.Set(reflect.ValueOf(resp.value))
 			return ok
 		case <-ctx.Done():
 			c.err = ctx.Err()

pool/cursor_test.go

@@ -10,27 +10,41 @@ import (
 )
 
 func TestCursor_Next(t *testing.T) {
-	c := Cursor{ch: make(chan response, 3)}
+
+	type structExample struct {
+		k1 int
+		k2 string
+		k3 []bool
+	}
+
+	c := Cursor{ch: make(chan response, 4)}
 
 	c.ch <- response{value: "12345"}
 	c.ch <- response{value: "abc"}
-	c.ch <- response{value: "xyz 0987"}
+	c.ch <- response{value: 1234.567}
+	c.ch <- response{value: structExample{k1: 12, k2: "abcd", k3: []bool{true, false}}}
 	close(c.ch)
 
-	var v interface{}
-	next := c.Next(context.Background(), &v)
+	var s string
+	next := c.Next(context.Background(), &s)
+	assert.True(t, next)
+	assert.Equal(t, "12345", s)
+
+	next = c.Next(context.Background(), &s)
 	assert.True(t, next)
-	assert.Equal(t, "12345", v.(string))
+	assert.Equal(t, "abc", s)
 
-	next = c.Next(context.Background(), &v)
+	var f float64
+	next = c.Next(context.Background(), &f)
 	assert.True(t, next)
-	assert.Equal(t, "abc", v.(string))
+	assert.Equal(t, 1234.567, f)
 
-	next = c.Next(context.Background(), &v)
+	var ss structExample
+	next = c.Next(context.Background(), &ss)
 	assert.True(t, next)
-	assert.Equal(t, "xyz 0987", v.(string))
+	assert.Equal(t, structExample{k1: 12, k2: "abcd", k3: []bool{true, false}}, ss)
 
-	next = c.Next(context.Background(), &v)
+	next = c.Next(context.Background(), nil)
 	assert.False(t, next)
 }
 

pool/pool_test.go

@@ -119,6 +119,45 @@ func TestPool(t *testing.T) {
 	}
 }
 
+func TestPoolWithStruct(t *testing.T) {
+
+	type resp struct {
+		k1 string
+		k2 int
+		k3 string
+		k4 []string
+		k5 bool
+	}
+
+	p := New(4, func(ctx context.Context, inpRec interface{}, sender SenderFn, store WorkerStore) error {
+		i := inpRec.(int)
+		r := resp{k1: "rec" + strconv.Itoa(i), k2: i, k3: "something", k4: []string{"foo", "bar"}, k5: true}
+		return sender(r)
+	})
+
+	go func() {
+		for i := 0; i < 1000; i++ {
+			p.Submit(i)
+			time.Sleep(time.Millisecond * time.Duration(rand.Intn(3))) //nolint gosec
+		}
+		p.Close()
+	}()
+	c, err := p.Go(context.Background())
+	require.NoError(t, err)
+
+	var recs []resp
+	var v resp
+	for c.Next(context.Background(), &v) {
+		recs = append(recs, v)
+	}
+	require.NoError(t, c.Err())
+	assert.Equal(t, 1000, len(recs))
+
+	sort.Slice(recs, func(i, j int) bool { return recs[i].k2 < recs[j].k2 })
+	assert.Equal(t, resp{k1: "rec0", k2: 0, k3: "something", k4: []string{"foo", "bar"}, k5: true}, recs[0])
+	assert.Equal(t, resp{k1: "rec999", k2: 999, k3: "something", k4: []string{"foo", "bar"}, k5: true}, recs[999])
+}
+
 func TestPoolWithStore(t *testing.T) {
 
 	worker := func(ctx context.Context, v interface{}, send SenderFn, store WorkerStore) error {
@@ -393,9 +432,9 @@ func ExampleWorkers_withOptions() {
 	}()
 
 	// consume results in streaming mode
-	var v interface{}
+	var v string
 	for cursor.Next(context.TODO(), &v) {
-		log.Printf("%v", v)
+		log.Printf("%s", v)
 	}
 
 	// show metrics