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

Make sure gofmt run on all files

Browse source
This commit is contained in:
Harlow Ward 2020-07-21 20:31:38 -07:00
parent 3b95644d77
commit bae065cf53
2 changed files with 31 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
store/memory/store.go
View file

@ -4,30 +4,30 @@
package store package store
import ( import (
"fmt" "fmt"
"sync" "sync"
) )
func New() *Store { func New() *Store {
return &Store{} return &Store{}
} }
type Store struct { type Store struct {
sync.Map sync.Map
} }
func (c *Store) SetCheckpoint(streamName, shardID, sequenceNumber string) error { func (c *Store) SetCheckpoint(streamName, shardID, sequenceNumber string) error {
if sequenceNumber == "" { if sequenceNumber == "" {
return fmt.Errorf("sequence number should not be empty") return fmt.Errorf("sequence number should not be empty")
} }
c.Store(streamName+":"+shardID, sequenceNumber) c.Store(streamName+":"+shardID, sequenceNumber)
return nil return nil
} }
func (c *Store) GetCheckpoint(streamName, shardID string) (string, error) { func (c *Store) GetCheckpoint(streamName, shardID string) (string, error) {
val, ok := c.Load(streamName + ":" + shardID) val, ok := c.Load(streamName + ":" + shardID)
if !ok { if !ok {
return "", nil return "", nil
} }
return val.(string), nil return val.(string), nil
} }

34
store/memory/store_test.go
View file

@ -1,30 +1,30 @@
package store package store
import ( import (
"testing" "testing"
) )
func Test_CheckpointLifecycle(t *testing.T) { func Test_CheckpointLifecycle(t *testing.T) {
c := New() c := New()
// set // set
c.SetCheckpoint("streamName", "shardID", "testSeqNum") c.SetCheckpoint("streamName", "shardID", "testSeqNum")
// get // get
val, err := c.GetCheckpoint("streamName", "shardID") val, err := c.GetCheckpoint("streamName", "shardID")
if err != nil { if err != nil {
t.Fatalf("get checkpoint error: %v", err) t.Fatalf("get checkpoint error: %v", err)
} }
if val != "testSeqNum" { if val != "testSeqNum" {
t.Fatalf("checkpoint exists expected %s, got %s", "testSeqNum", val) t.Fatalf("checkpoint exists expected %s, got %s", "testSeqNum", val)
} }
} }
func Test_SetEmptySeqNum(t *testing.T) { func Test_SetEmptySeqNum(t *testing.T) {
c := New() c := New()
err := c.SetCheckpoint("streamName", "shardID", "") err := c.SetCheckpoint("streamName", "shardID", "")
if err == nil || err.Error() != "sequence number should not be empty" { if err == nil || err.Error() != "sequence number should not be empty" {
t.Fatalf("should not allow empty sequence number") t.Fatalf("should not allow empty sequence number")
} }
} }