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

#188 allows redis store to use clustered redis

Browse source
This commit is contained in:
Alex Senger 2024-09-13 14:14:56 +02:00
parent 3775a5af04
commit 1760aaf19d
No known key found for this signature in database
GPG key ID: 0B4A96F8AF6934CF
6 changed files with 49 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
examples/consumer-postgres/main.go
View file

@ -4,7 +4,7 @@ import (
"context"
"flag"
"fmt"
"log"
"log/slog"
"os"
"os/signal"
@ -16,45 +16,50 @@ import (
store "github.com/alexgridx/kinesis-consumer/store/postgres"
)
var (
applicationName = flag.String("application.name", "", "Consumer app name")
kinesisAWSRegion = flag.String("kinesis.region", "us-west-2", "AWS Region")
kinesisEndpoint = flag.String("kinesis.endpoint", "http://localhost:4567", "Kinesis endpoint")
kinesisStream = flag.String("kinesis.stream", "", "Stream name")
postgresConnection = flag.String("postgres.connection", "", "Connection Str")
postgresTable = flag.String("postgres.table", "", "Table name")
)
func main() {
var (
app = flag.String("app", "", "Consumer app name")
stream = flag.String("stream", "", "Stream name")
table = flag.String("table", "", "Table name")
connStr = flag.String("connection", "", "Connection Str")
kinesisEndpoint = flag.String("endpoint", "http://localhost:4567", "Kinesis endpoint")
awsRegion = flag.String("region", "us-west-2", "AWS Region")
)
flag.Parse()
// postgres checkpoint
checkpointStore, err := store.New(*app, *table, *connStr)
checkpointStore, err := store.New(*applicationName, *postgresTable, *postgresConnection)
if err != nil {
log.Fatalf("checkpoint error: %v", err)
slog.Error("checkpoint error", slog.String("error", err.Error()))
os.Exit(1)
}
registry, ok := prometheus.DefaultRegisterer.(*prometheus.Registry)
if !ok {
log.Fatalf("could not get default prometheus registry")
slog.Error("prometheus error")
os.Exit(1)
}
// client
var client = kinesis.New(
kinesis.Options{
BaseEndpoint: kinesisEndpoint,
Region: *awsRegion,
Region: *kinesisAWSRegion,
Credentials: credentials.NewStaticCredentialsProvider("user", "pass", "token"),
})
// consumer
c, err := consumer.New(
*stream,
*kinesisStream,
consumer.WithClient(client),
consumer.WithStore(checkpointStore),
consumer.WithMetricRegistry(registry),
consumer.WithLogger(slog.Default()),
)
if err != nil {
log.Fatalf("consumer error: %v", err)
slog.Error("consumer error", slog.String("error", err.Error()))
os.Exit(1)
}
// use cancel func to signal shutdown
@ -76,10 +81,12 @@ func main() {
})
if err != nil {
log.Fatalf("scan error: %v", err)
slog.Error("scan error", slog.String("error", err.Error()))
os.Exit(1)
}
if err := checkpointStore.Shutdown(); err != nil {
log.Fatalf("store shutdown error: %v", err)
slog.Error("store shutdown error", slog.String("error", err.Error()))
os.Exit(1)
}
}

45
examples/consumer-redis/main.go
View file

@ -4,7 +4,7 @@ import (
"context"
"flag"
"fmt"
"log"
"log/slog"
"os"
"os/signal"
@ -15,53 +15,41 @@ import (
store "github.com/alexgridx/kinesis-consumer/store/redis"
)
// A myLogger provides a minimalistic logger satisfying the Logger interface.
type myLogger struct {
logger *log.Logger
}
// Log logs the parameters to the stdlib logger. See log.Println.
func (l *myLogger) Log(args ...interface{}) {
l.logger.Println(args...)
}
var (
applicationName = flag.String("application.name", "", "Consumer app name")
kinesisAWSRegion = flag.String("kinesis.region", "us-west-2", "AWS Region")
kinesisEndpoint = flag.String("kinesis.endpoint", "http://localhost:4567", "Kinesis endpoint")
kinesisStream = flag.String("kinesis.stream", "", "Stream name")
)
func main() {
var (
app = flag.String("app", "", "Consumer app name")
stream = flag.String("stream", "", "Stream name")
kinesisEndpoint = flag.String("endpoint", "http://localhost:4567", "Kinesis endpoint")
awsRegion = flag.String("region", "us-west-2", "AWS Region")
)
flag.Parse()
// redis checkpoint checkpointStore
checkpointStore, err := store.New(*app)
checkpointStore, err := store.New(*applicationName)
if err != nil {
log.Fatalf("checkpointStore error: %v", err)
}
// logger
logger := &myLogger{
logger: log.New(os.Stdout, "consumer-example: ", log.LstdFlags),
slog.Error("checkpoint store error", slog.String("error", err.Error()))
os.Exit(1)
}
// client
var client = kinesis.New(
kinesis.Options{
BaseEndpoint: kinesisEndpoint,
Region: *awsRegion,
Region: *kinesisAWSRegion,
Credentials: credentials.NewStaticCredentialsProvider("user", "pass", "token"),
})
// consumer
c, err := consumer.New(
*stream,
*kinesisStream,
consumer.WithClient(client),
consumer.WithStore(checkpointStore),
consumer.WithLogger(logger),
consumer.WithLogger(slog.Default()),
)
if err != nil {
log.Fatalf("consumer error: %v", err)
slog.Error("consumer error", slog.String("error", err.Error()))
os.Exit(1)
}
// use cancel func to signal shutdown
@ -83,6 +71,7 @@ func main() {
return nil // continue scanning
})
if err != nil {
log.Fatalf("scan error: %v", err)
slog.Error("scan error", slog.String("error", err.Error()))
os.Exit(1)
}
}

2
store/postgres/postgres_test.go
View file

@ -1,3 +1,5 @@
//go:build unit
package postgres
import (

2
store/redis/options.go
View file

@ -6,7 +6,7 @@ import "github.com/redis/go-redis/v9"
type Option func(*Checkpoint)
// WithClient overrides the default client
func WithClient(client *redis.Client) Option {
func WithClient(client redis.UniversalClient) Option {
return func(c *Checkpoint) {
c.client = client
}

6
store/redis/redis.go
View file

@ -45,10 +45,10 @@ func New(appName string, opts ...Option) (*Checkpoint, error) {
return c, nil
}
// Checkpoint stores and retreives the last evaluated key from a DDB scan
// Checkpoint stores and retrieves the last evaluated key from a DDB scan
type Checkpoint struct {
appName string
client *redis.Client
client redis.UniversalClient
}
// GetCheckpoint fetches the checkpoint for a particular Shard.
@ -59,7 +59,7 @@ func (c *Checkpoint) GetCheckpoint(streamName, shardID string) (string, error) {
}
// SetCheckpoint stores a checkpoint for a shard (e.g. sequence number of last record processed by application).
// Upon failover, record processing is resumed from this point.
// Upon fail over, record processing is resumed from this point.
func (c *Checkpoint) SetCheckpoint(streamName, shardID, sequenceNumber string) error {
if sequenceNumber == "" {
return fmt.Errorf("sequence number should not be empty")

2
store/redis/redis_test.go
View file

@ -1,3 +1,5 @@
// go:build unit
package redis
import (