| 78 | } |
| 79 | |
| 80 | func TestAddTask(t *testing.T) { |
| 81 | a, ctx := test.New(t) |
| 82 | |
| 83 | cl, flush := test.NewRedis(ctx, "redis_test") |
| 84 | defer flush() |
| 85 | defer cl.Close() |
| 86 | |
| 87 | err := AddTask(ctx, cl, cl.Key("testKey"), 10, "testPayload", time.Unix(0, 42), false) |
| 88 | if !a.So(err, should.BeNil) { |
| 89 | t.FailNow() |
| 90 | } |
| 91 | |
| 92 | rets, err := cl.Client.XRead(ctx, &redis.XReadArgs{ |
| 93 | Streams: []string{InputTaskKey(cl.Key("testKey")), "0"}, |
| 94 | Count: 10, |
| 95 | Block: -1, |
| 96 | }).Result() |
| 97 | if !a.So(err, should.BeNil) { |
| 98 | t.FailNow() |
| 99 | } |
| 100 | |
| 101 | if a.So(rets, should.HaveLength, 1) { |
| 102 | a.So(rets[0].Stream, should.Equal, InputTaskKey(cl.Key("testKey"))) |
| 103 | if a.So(rets[0].Messages, should.HaveLength, 1) { |
| 104 | msg := rets[0].Messages[0] |
| 105 | a.So(msg, should.Resemble, redis.XMessage{ |
| 106 | ID: msg.ID, |
| 107 | Values: map[string]any{ |
| 108 | "start_at": fmt.Sprintf("%d", time.Unix(0, 42).UnixNano()), |
| 109 | "payload": "testPayload", |
| 110 | }, |
| 111 | }) |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | err = AddTask(ctx, cl, cl.Key("testKey"), 10, "testPayload", time.Unix(0, 42), true) |
| 116 | if !a.So(err, should.BeNil) { |
| 117 | t.FailNow() |
| 118 | } |
| 119 | |
| 120 | rets, err = cl.Client.XRead(ctx, &redis.XReadArgs{ |
| 121 | Streams: []string{InputTaskKey(cl.Key("testKey")), "0"}, |
| 122 | Count: 10, |
| 123 | Block: -1, |
| 124 | }).Result() |
| 125 | if !a.So(err, should.BeNil) { |
| 126 | t.FailNow() |
| 127 | } |
| 128 | |
| 129 | if a.So(rets, should.HaveLength, 1) { |
| 130 | a.So(rets[0].Stream, should.Equal, InputTaskKey(cl.Key("testKey"))) |
| 131 | if a.So(rets[0].Messages, should.HaveLength, 2) { |
| 132 | msg0 := rets[0].Messages[0] |
| 133 | a.So(msg0, should.Resemble, redis.XMessage{ |
| 134 | ID: msg0.ID, |
| 135 | Values: map[string]any{ |
| 136 | "start_at": fmt.Sprintf("%d", time.Unix(0, 42).UnixNano()), |
| 137 | "payload": "testPayload", |