test upgrade

basic_test.go

@@ -11,6 +11,13 @@ import (
 )
 
 func TestBasicUsage(t *testing.T) {
+	// consumer group must be created before posting messages with unattached consumers
+	if _, err := redisCli.XGroupCreateMkStream(context.Background(),
+		"topics/1", "group1", "$").Result(); err != nil {
+		t.Error(err)
+		return
+	}
+
 	ctx := context.Background()
 	topic, err := pubsub.OpenTopic(ctx, "redis://topics/1")
 	if err != nil {
@@ -30,6 +37,13 @@ func TestBasicUsage(t *testing.T) {
 		},
 	}
 
+	// send before consumer attach
+	err = topic.Send(ctx, orig)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 
@@ -41,12 +55,12 @@ func TestBasicUsage(t *testing.T) {
 			return
 		}
 		defer subs.Shutdown(ctx)
-		for {
+		for i := 0; i < 2; i++ {
 			msg, err := subs.Receive(ctx)
 			if err != nil {
 				// Errors from Receive indicate that Receive will no longer succeed.
 				t.Errorf("Receiving message: %v", err)
-				break
+				return
 			}
 			// Do work based on the message, for example:
 			t.Logf("Got message: %q\n", msg.Body)
@@ -55,22 +69,29 @@ func TestBasicUsage(t *testing.T) {
 
 			if !bytes.Equal(msg.Body, orig.Body) {
 				t.Error("body not equal")
+				return
 			}
 			for k, v := range msg.Metadata {
 				if orig.Metadata[k] != v {
 					t.Error("metadata not equal")
+					return
+				}
+			}
+			if i == 0 {
+				// send after consumer attached
+				err = topic.Send(ctx, orig)
+				if err != nil {
+					t.Error(err)
+					return
 				}
 			}
-
-			break
 		}
 	}()
 
-	err = topic.Send(ctx, orig)
-	if err != nil {
-		t.Error(err)
-		return
-	}
-
 	wg.Wait()
+
+	res, err := redisCli.XPending(ctx, "topics/1", "group1").Result()
+	if res.Count != 0 {
+		t.Error(res.Count, err)
+	}
 }

main_test.go

@@ -13,6 +13,8 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+var redisCli *redis.Client
+
 func TestMain(m *testing.M) {
 	// uses a sensible default on windows (tcp/http) and linux/osx (socket)
 	pool, err := dockertest.NewPool("")
@@ -49,6 +51,7 @@ func TestMain(m *testing.M) {
 			return err
 		}
 		rdb := redis.NewClient(opt)
+		redisCli = rdb
 		_, err = rdb.Ping(context.Background()).Result()
 		return err
 	}); err != nil {