luciano/kinesis-consumer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Specify stop scan when returning scan status

Browse source
This commit is contained in:
Harlow Ward 2018-12-28 19:34:13 -08:00
parent c061203d5b
commit 5688ff2820
1 changed files with 10 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
consumer_test.go
View file

@ -12,8 +12,7 @@ import (
) )
func TestNew(t *testing.T) { func TestNew(t *testing.T) {
_, err := New("myStreamName") if _, err := New("myStreamName"); err != nil {
if err != nil {
t.Fatalf("new consumer error: %v", err) t.Fatalf("new consumer error: %v", err)
} }
} }
@ -180,9 +179,9 @@ func TestScanShard(t *testing.T) {
} }
// callback fn appends record data // callback fn appends record data
var resultData string var res string
var fn = func(r *Record) ScanStatus { var fn = func(r *Record) ScanStatus {
resultData += string(r.Data) res += string(r.Data)
return ScanStatus{} return ScanStatus{}
} }
@ -192,8 +191,8 @@ func TestScanShard(t *testing.T) {
} }
// runs callback func // runs callback func
if resultData != "firstDatalastData" { if res != "firstDatalastData" {
t.Fatalf("callback error expected %s, got %s", "firstDatalastData", resultData) t.Fatalf("callback error expected %s, got %s", "firstDatalastData", res)
} }
// increments counter // increments counter
@ -240,9 +239,9 @@ func TestScanShard_StopScan(t *testing.T) {
} }
// callback fn appends record data // callback fn appends record data
var resultData string var res string
var fn = func(r *Record) ScanStatus { var fn = func(r *Record) ScanStatus {
resultData += string(r.Data) res += string(r.Data)
return ScanStatus{StopScan: true} return ScanStatus{StopScan: true}
} }
@ -250,8 +249,8 @@ func TestScanShard_StopScan(t *testing.T) {
t.Fatalf("scan shard error: %v", err) t.Fatalf("scan shard error: %v", err)
} }
if resultData != "firstData" { if res != "firstData" {
t.Fatalf("callback error expected %s, got %s", "firstData", resultData) t.Fatalf("callback error expected %s, got %s", "firstData", res)
} }
} }
@ -276,7 +275,7 @@ func TestScanShard_ShardIsClosed(t *testing.T) {
} }
var fn = func(r *Record) ScanStatus { var fn = func(r *Record) ScanStatus {
return ScanStatus{} return ScanStatus{StopScan: true}
} }
if err := c.ScanShard(context.Background(), "myShard", fn); err != nil { if err := c.ScanShard(context.Background(), "myShard", fn); err != nil {