diff --git a/parser/cmd/main.go b/parser/cmd/main.go index 20c05714..551658e0 100644 --- a/parser/cmd/main.go +++ b/parser/cmd/main.go @@ -7,12 +7,12 @@ import ( "os/signal" "syscall" - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/credentials" - "github.com/aws/aws-sdk-go/aws/session" + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" ipfslog "github.com/ipfs/go-log/v2" "github.com/wormhole-foundation/wormhole-explorer/parser/config" - "github.com/wormhole-foundation/wormhole-explorer/parser/http/infraestructure" + "github.com/wormhole-foundation/wormhole-explorer/parser/http/infrastructure" "github.com/wormhole-foundation/wormhole-explorer/parser/internal/db" "github.com/wormhole-foundation/wormhole-explorer/parser/internal/sqs" "github.com/wormhole-foundation/wormhole-explorer/parser/parser" @@ -65,7 +65,7 @@ func main() { } // get publish function. - sqsConsumer, vaaPushFunc, vaaConsumeFunc := newVAAPublishAndConsume(config, logger) + sqsConsumer, vaaPushFunc, vaaConsumeFunc := newVAAPublishAndConsume(rootCtx, config, logger) repository := parser.NewRepository(db.Database, logger) // // create a new publisher. @@ -80,7 +80,7 @@ func main() { consumer := pipeline.NewConsumer(vaaConsumeFunc, repository, parserVAAAPIClient, logger) consumer.Start(rootCtx) - server := infraestructure.NewServer(logger, config.Port, config.PprofEnabled, config.IsQueueConsumer(), sqsConsumer, db.Database) + server := infrastructure.NewServer(logger, config.Port, config.PprofEnabled, config.IsQueueConsumer(), sqsConsumer, db.Database) server.Start() logger.Info("Started wormhole-explorer-parser") @@ -105,32 +105,43 @@ func main() { logger.Info("Finished wormhole-explorer-parser") } -func newAwsSession(cfg *config.Configuration) (*session.Session, error) { +func newAwsConfig(appCtx context.Context, cfg *config.Configuration) (aws.Config, error) { region := cfg.AwsRegion - config := aws.NewConfig().WithRegion(region) - if cfg.AwsAccessKeyID != "" && cfg.AwsSecretAccessKey != "" { - config.WithCredentials(credentials.NewStaticCredentials(cfg.AwsAccessKeyID, cfg.AwsSecretAccessKey, "")) - } - if cfg.AwsEndpoint != "" { - config.WithEndpoint(cfg.AwsEndpoint) - } - return session.NewSession(config) + credentials := credentials.NewStaticCredentialsProvider(cfg.AwsAccessKeyID, cfg.AwsSecretAccessKey, "") + customResolver := aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) { + if cfg.AwsEndpoint != "" { + return aws.Endpoint{ + PartitionID: "aws", + URL: cfg.AwsEndpoint, + SigningRegion: region, + }, nil + } + + return aws.Endpoint{}, &aws.EndpointNotFoundError{} + }) + + awsCfg, err := awsconfig.LoadDefaultConfig(appCtx, + awsconfig.WithRegion(region), + awsconfig.WithEndpointResolver(customResolver), + awsconfig.WithCredentialsProvider(credentials), + ) + return awsCfg, err } // Creates two callbacks depending on whether the execution is local (memory queue) or not (SQS queue) -func newVAAPublishAndConsume(config *config.Configuration, logger *zap.Logger) (*sqs.Consumer, queue.VAAPushFunc, queue.VAAConsumeFunc) { +func newVAAPublishAndConsume(appCtx context.Context, config *config.Configuration, logger *zap.Logger) (*sqs.Consumer, queue.VAAPushFunc, queue.VAAConsumeFunc) { // check is consumer type. if !config.IsQueueConsumer() { vaaQueue := queue.NewVAAInMemory() return nil, vaaQueue.Publish, vaaQueue.Consume } - sqsConsumer, err := newSQSConsumer(config) + sqsConsumer, err := newSQSConsumer(appCtx, config) if err != nil { logger.Fatal("failed to create sqs consumer", zap.Error(err)) } - sqsProducer, err := newSQSProducer(config) + sqsProducer, err := newSQSProducer(appCtx, config) if err != nil { logger.Fatal("failed to create sqs producer", zap.Error(err)) } @@ -139,22 +150,22 @@ func newVAAPublishAndConsume(config *config.Configuration, logger *zap.Logger) ( return sqsConsumer, vaaQueue.Publish, vaaQueue.Consume } -func newSQSProducer(config *config.Configuration) (*sqs.Producer, error) { - session, err := newAwsSession(config) +func newSQSProducer(appCtx context.Context, config *config.Configuration) (*sqs.Producer, error) { + awsConfig, err := newAwsConfig(appCtx, config) if err != nil { return nil, err } - return sqs.NewProducer(session, config.SQSUrl) + return sqs.NewProducer(awsConfig, config.SQSUrl) } -func newSQSConsumer(config *config.Configuration) (*sqs.Consumer, error) { - session, err := newAwsSession(config) +func newSQSConsumer(appCtx context.Context, config *config.Configuration) (*sqs.Consumer, error) { + awsconfig, err := newAwsConfig(appCtx, config) if err != nil { return nil, err } - return sqs.NewConsumer(session, config.SQSUrl, + return sqs.NewConsumer(awsconfig, config.SQSUrl, sqs.WithMaxMessages(10), sqs.WithVisibilityTimeout(120)) } diff --git a/parser/go.mod b/parser/go.mod index c2cfe818..12725aae 100644 --- a/parser/go.mod +++ b/parser/go.mod @@ -3,7 +3,6 @@ module github.com/wormhole-foundation/wormhole-explorer/parser go 1.19 require ( - github.com/aws/aws-sdk-go v1.44.161 github.com/gofiber/fiber/v2 v2.40.1 github.com/ipfs/go-log/v2 v2.5.1 github.com/joho/godotenv v1.4.0 // Configuration environment @@ -15,15 +14,29 @@ require ( go.uber.org/zap v1.23.0 ) +require ( + github.com/aws/aws-sdk-go-v2 v1.17.3 + github.com/aws/aws-sdk-go-v2/config v1.1.1 + github.com/aws/aws-sdk-go-v2/credentials v1.1.1 + github.com/aws/aws-sdk-go-v2/service/sqs v1.20.1 +) + require ( github.com/andybalholm/brotli v1.0.4 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.0.2 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.27 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.21 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.3.28 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.0.2 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.1.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.1.1 // indirect + github.com/aws/smithy-go v1.13.5 // indirect github.com/benbjohnson/clock v1.3.0 // indirect github.com/btcsuite/btcd/btcec/v2 v2.2.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 // indirect github.com/ethereum/go-ethereum v1.10.21 // indirect github.com/golang/snappy v0.0.4 // indirect github.com/google/go-cmp v0.5.9 // indirect - github.com/jmespath/go-jmespath v0.4.0 // indirect github.com/klauspost/compress v1.15.11 // indirect github.com/kr/pretty v0.3.0 // indirect github.com/mattn/go-colorable v0.1.13 // indirect diff --git a/parser/go.sum b/parser/go.sum index ae38e78e..f89dbbc1 100644 --- a/parser/go.sum +++ b/parser/go.sum @@ -1,7 +1,31 @@ github.com/andybalholm/brotli v1.0.4 h1:V7DdXeJtZscaqfNuAdSRuRFzuiKlHSC/Zh3zl9qY3JY= github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= -github.com/aws/aws-sdk-go v1.44.161 h1:uZdZJ30mlbaU2wsrd/wzibrX01cbgKE2t486TtRjeHs= -github.com/aws/aws-sdk-go v1.44.161/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI= +github.com/aws/aws-sdk-go-v2 v1.2.0/go.mod h1:zEQs02YRBw1DjK0PoJv3ygDYOFTre1ejlJWl8FwAuQo= +github.com/aws/aws-sdk-go-v2 v1.17.3 h1:shN7NlnVzvDUgPQ+1rLMSxY8OWRNDRYtiqe0p/PgrhY= +github.com/aws/aws-sdk-go-v2 v1.17.3/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw= +github.com/aws/aws-sdk-go-v2/config v1.1.1 h1:ZAoq32boMzcaTW9bcUacBswAmHTbvlvDJICgHFZuECo= +github.com/aws/aws-sdk-go-v2/config v1.1.1/go.mod h1:0XsVy9lBI/BCXm+2Tuvt39YmdHwS5unDQmxZOYe8F5Y= +github.com/aws/aws-sdk-go-v2/credentials v1.1.1 h1:NbvWIM1Mx6sNPTxowHgS2ewXCRp+NGTzUYb/96FZJbY= +github.com/aws/aws-sdk-go-v2/credentials v1.1.1/go.mod h1:mM2iIjwl7LULWtS6JCACyInboHirisUUdkBPoTHMOUo= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.0.2 h1:EtEU7WRaWliitZh2nmuxEXrN0Cb8EgPUFGIoTMeqbzI= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.0.2/go.mod h1:3hGg3PpiEjHnrkrlasTfxFqUsZ2GCk/fMUn4CbKgSkM= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.27 h1:I3cakv2Uy1vNmmhRQmFptYDxOvBnwCdNwyw63N0RaRU= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.27/go.mod h1:a1/UpzeyBBerajpnP5nGZa9mGzsBn5cOKxm6NWQsvoI= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.21 h1:5NbbMrIzmUn/TXFqAle6mgrH5m9cOvMLRGL7pnG8tRE= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.21/go.mod h1:+Gxn8jYn5k9ebfHEqlhrMirFjSW0v0C9fI+KN5vk2kE= +github.com/aws/aws-sdk-go-v2/internal/ini v1.3.28 h1:KeTxcGdNnQudb46oOl4d90f2I33DF/c6q3RnZAmvQdQ= +github.com/aws/aws-sdk-go-v2/internal/ini v1.3.28/go.mod h1:yRZVr/iT0AqyHeep00SZ4YfBAKojXz08w3XMBscdi0c= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.0.2 h1:4AH9fFjUlVktQMznF+YN33aWNXaR4VgDXyP28qokJC0= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.0.2/go.mod h1:45MfaXZ0cNbeuT0KQ1XJylq8A6+OpVV2E5kvY/Kq+u8= +github.com/aws/aws-sdk-go-v2/service/sqs v1.20.1 h1:JvO+TT1JhH8InfwOfgWAfIFo3H1cz5qW8WuIP8Y5d6s= +github.com/aws/aws-sdk-go-v2/service/sqs v1.20.1/go.mod h1:jQhN5f4p3PALMNlUtfb/0wGIFlV7vGtJlPDVfxfNfPY= +github.com/aws/aws-sdk-go-v2/service/sso v1.1.1 h1:37QubsarExl5ZuCBlnRP+7l1tNwZPBSTqpTBrPH98RU= +github.com/aws/aws-sdk-go-v2/service/sso v1.1.1/go.mod h1:SuZJxklHxLAXgLTc1iFXbEWkXs7QRTQpCLGaKIprQW0= +github.com/aws/aws-sdk-go-v2/service/sts v1.1.1 h1:TJoIfnIFubCX0ACVeJ0w46HEH5MwjwYN4iFhuYIhfIY= +github.com/aws/aws-sdk-go-v2/service/sts v1.1.1/go.mod h1:Wi0EBZwiz/K44YliU0EKxqTCJGUfYTWXrrBwkq736bM= +github.com/aws/smithy-go v1.1.0/go.mod h1:EzMw8dbp/YJL4A5/sbhGddag+NPT7q084agLbB9LgIw= +github.com/aws/smithy-go v1.13.5 h1:hgz0X/DX0dGqTYpGALqXJoRKRj5oQ7150i5FdTePzO8= +github.com/aws/smithy-go v1.13.5/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA= github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A= github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= @@ -22,14 +46,15 @@ github.com/gofiber/fiber/v2 v2.40.1/go.mod h1:Gko04sLksnHbzLSRBFWPFdzM9Ws9pRxvvI github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/ipfs/go-log/v2 v2.5.1 h1:1XdUzF7048prq4aBjDQQ4SL5RxftpRGdXhNRwKSAlcY= github.com/ipfs/go-log/v2 v2.5.1/go.mod h1:prSpmC1Gpllc9UYWxDiZDreBYw7zp4Iqp1kOLU9U5UI= -github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg= github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= -github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U= github.com/joho/godotenv v1.4.0 h1:3l4+N6zfMWnkbPEXKng2o2/MR5mSwTrBih4ZEkkz1lg= github.com/joho/godotenv v1.4.0/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= @@ -94,7 +119,6 @@ github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgk github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA= github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= -github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.mongodb.org/mongo-driver v1.11.0 h1:FZKhBSTydeuffHj9CBjXlR8vQLee1cQyTWYPA6/tqiE= go.mongodb.org/mongo-driver v1.11.0/go.mod h1:s7p5vEtfbeR1gYi6pnj3c3/urpbLv2T5Sfd6Rp2HBB8= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= @@ -111,26 +135,20 @@ go.uber.org/zap v1.23.0 h1:OjGQ5KQDEUawVHxNwQgPpiypGHOxo2mNZsOqTak4fFY= go.uber.org/zap v1.23.0/go.mod h1:D+nX8jyLsMHMYrln8A0rJjFt/T/9/bGgIhAqxv5URuY= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= -golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20220214200702-86341886e292/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU= golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw= golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220906165146-f3363e06e74c/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk= -golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -141,16 +159,12 @@ golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A= golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= @@ -161,7 +175,6 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= -golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -170,7 +183,6 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/parser/http/infraestructure/controller.go b/parser/http/infrastructure/controller.go similarity index 97% rename from parser/http/infraestructure/controller.go rename to parser/http/infrastructure/controller.go index 793bb4f7..44540c2a 100644 --- a/parser/http/infraestructure/controller.go +++ b/parser/http/infrastructure/controller.go @@ -1,4 +1,4 @@ -package infraestructure +package infrastructure import "github.com/gofiber/fiber/v2" diff --git a/parser/http/infraestructure/model.go b/parser/http/infrastructure/model.go similarity index 96% rename from parser/http/infraestructure/model.go rename to parser/http/infrastructure/model.go index c96ead64..4cbc1b01 100644 --- a/parser/http/infraestructure/model.go +++ b/parser/http/infrastructure/model.go @@ -1,4 +1,4 @@ -package infraestructure +package infrastructure // MongoStatus represent a mongo server status. type MongoStatus struct { diff --git a/parser/http/infraestructure/repository.go b/parser/http/infrastructure/repository.go similarity index 98% rename from parser/http/infraestructure/repository.go rename to parser/http/infrastructure/repository.go index f3417e17..911d984d 100644 --- a/parser/http/infraestructure/repository.go +++ b/parser/http/infrastructure/repository.go @@ -1,4 +1,4 @@ -package infraestructure +package infrastructure import ( "context" diff --git a/parser/http/infraestructure/server.go b/parser/http/infrastructure/server.go similarity index 98% rename from parser/http/infraestructure/server.go rename to parser/http/infrastructure/server.go index 08eb2af4..9c840719 100644 --- a/parser/http/infraestructure/server.go +++ b/parser/http/infrastructure/server.go @@ -1,4 +1,4 @@ -package infraestructure +package infrastructure import ( "github.com/gofiber/fiber/v2" diff --git a/parser/http/infraestructure/service.go b/parser/http/infrastructure/service.go similarity index 96% rename from parser/http/infraestructure/service.go rename to parser/http/infrastructure/service.go index 6b7c9a0a..e1470a54 100644 --- a/parser/http/infraestructure/service.go +++ b/parser/http/infrastructure/service.go @@ -1,4 +1,4 @@ -package infraestructure +package infrastructure import ( "context" @@ -76,8 +76,8 @@ func (s *Service) CheckAwsSQS(ctx context.Context) (bool, error) { // check queue created createdTimestamp := queueAttributes.Attributes["CreatedTimestamp"] - if createdTimestamp == nil { + if createdTimestamp == "" { return false, errors.New("error createdTimestamp attributes does not exist") } - return *createdTimestamp != "", nil + return createdTimestamp != "", nil } diff --git a/parser/internal/sqs/sqs_consumer.go b/parser/internal/sqs/sqs_consumer.go index acb7adbf..cca2ab06 100644 --- a/parser/internal/sqs/sqs_consumer.go +++ b/parser/internal/sqs/sqs_consumer.go @@ -1,12 +1,12 @@ package sqs import ( + "context" "time" - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/session" - aws_sqs "github.com/aws/aws-sdk-go/service/sqs" - "github.com/aws/aws-sdk-go/service/sqs/sqsiface" + "github.com/aws/aws-sdk-go-v2/aws" + aws_sqs "github.com/aws/aws-sdk-go-v2/service/sqs" + aws_sqs_types "github.com/aws/aws-sdk-go-v2/service/sqs/types" ) // ConsumerOption represents a consumer option function. @@ -14,21 +14,21 @@ type ConsumerOption func(*Consumer) // Consumer represents SQS consumer. type Consumer struct { - api sqsiface.SQSAPI + api *aws_sqs.Client url string - maxMessages *int64 - visibilityTimeout *int64 - waitTimeSeconds *int64 + maxMessages int32 + visibilityTimeout int32 + waitTimeSeconds int32 } // New instances of a Consumer to consume SQS messages. -func NewConsumer(sess *session.Session, url string, opts ...ConsumerOption) (*Consumer, error) { +func NewConsumer(awsConfig aws.Config, url string, opts ...ConsumerOption) (*Consumer, error) { consumer := &Consumer{ - api: aws_sqs.New(sess), + api: aws_sqs.NewFromConfig(awsConfig), url: url, - maxMessages: aws.Int64(10), - visibilityTimeout: aws.Int64(60), - waitTimeSeconds: aws.Int64(20), + maxMessages: 10, + visibilityTimeout: 60, + waitTimeSeconds: 20, } for _, opt := range opts { @@ -39,42 +39,42 @@ func NewConsumer(sess *session.Session, url string, opts ...ConsumerOption) (*Co } // WithMaxMessages allows to specify an maximum number of messages to return when setting a value. -func WithMaxMessages(v int64) ConsumerOption { +func WithMaxMessages(v int32) ConsumerOption { return func(c *Consumer) { - c.maxMessages = aws.Int64(v) + c.maxMessages = v } } // WithVisibilityTimeout allows to specify a visibility timeout when setting a value. -func WithVisibilityTimeout(v int64) ConsumerOption { +func WithVisibilityTimeout(v int32) ConsumerOption { return func(c *Consumer) { - c.visibilityTimeout = aws.Int64(v) + c.visibilityTimeout = v } } // WithWaitTimeSeconds allows to specify a wait time when setting a value. -func WithWaitTimeSeconds(v int64) ConsumerOption { +func WithWaitTimeSeconds(v int32) ConsumerOption { return func(c *Consumer) { - c.waitTimeSeconds = aws.Int64(v) + c.waitTimeSeconds = v } } // GetMessages retrieves messages from SQS. -func (c *Consumer) GetMessages() ([]*aws_sqs.Message, error) { +func (c *Consumer) GetMessages() ([]aws_sqs_types.Message, error) { params := &aws_sqs.ReceiveMessageInput{ QueueUrl: aws.String(c.url), MaxNumberOfMessages: c.maxMessages, - AttributeNames: []*string{ - aws.String("All"), + AttributeNames: []aws_sqs_types.QueueAttributeName{ + aws_sqs_types.QueueAttributeNameAll, }, - MessageAttributeNames: []*string{ - aws.String("All"), + MessageAttributeNames: []string{ + string(aws_sqs_types.QueueAttributeNameAll), }, WaitTimeSeconds: c.waitTimeSeconds, VisibilityTimeout: c.visibilityTimeout, } - res, err := c.api.ReceiveMessage(params) + res, err := c.api.ReceiveMessage(context.TODO(), params) if err != nil { return nil, err } @@ -83,28 +83,28 @@ func (c *Consumer) GetMessages() ([]*aws_sqs.Message, error) { } // DeleteMessage deletes messages from SQS. -func (c *Consumer) DeleteMessage(msg *aws_sqs.Message) error { +func (c *Consumer) DeleteMessage(msg *aws_sqs_types.Message) error { params := &aws_sqs.DeleteMessageInput{ QueueUrl: aws.String(c.url), ReceiptHandle: msg.ReceiptHandle, } - _, err := c.api.DeleteMessage(params) + _, err := c.api.DeleteMessage(context.TODO(), params) return err } // GetVisibilityTimeout returns visibility timeout. func (c *Consumer) GetVisibilityTimeout() time.Duration { - return time.Duration(*c.visibilityTimeout * int64(time.Second)) + return time.Duration(int64(c.visibilityTimeout) * int64(time.Second)) } // GetQueueAttributes get queue attributes. func (c *Consumer) GetQueueAttributes() (*aws_sqs.GetQueueAttributesOutput, error) { params := &aws_sqs.GetQueueAttributesInput{ QueueUrl: aws.String(c.url), - AttributeNames: []*string{ - aws.String("CreatedTimestamp"), + AttributeNames: []aws_sqs_types.QueueAttributeName{ + aws_sqs_types.QueueAttributeNameCreatedTimestamp, }, } - return c.api.GetQueueAttributes(params) + return c.api.GetQueueAttributes(context.TODO(), params) } diff --git a/parser/internal/sqs/sqs_producer.go b/parser/internal/sqs/sqs_producer.go index dd2e6840..0d9ed122 100644 --- a/parser/internal/sqs/sqs_producer.go +++ b/parser/internal/sqs/sqs_producer.go @@ -1,29 +1,28 @@ package sqs import ( - "github.com/aws/aws-sdk-go/aws" - "github.com/aws/aws-sdk-go/aws/session" - aws_sqs "github.com/aws/aws-sdk-go/service/sqs" - "github.com/aws/aws-sdk-go/service/sqs/sqsiface" + "context" + + "github.com/aws/aws-sdk-go-v2/aws" + aws_sqs "github.com/aws/aws-sdk-go-v2/service/sqs" ) // Producer represents SQS producer. type Producer struct { - api sqsiface.SQSAPI + api *aws_sqs.Client url string } -// NewProducer create a new instance of Producer. -func NewProducer(sess *session.Session, url string) (*Producer, error) { +func NewProducer(awsConfig aws.Config, url string) (*Producer, error) { return &Producer{ - api: aws_sqs.New(sess), + api: aws_sqs.NewFromConfig(awsConfig), url: url, }, nil } // SendMessage sends messages to SQS. func (p *Producer) SendMessage(groupID, deduplicationID, body string) error { - _, err := p.api.SendMessage( + _, err := p.api.SendMessage(context.TODO(), &aws_sqs.SendMessageInput{ MessageGroupId: aws.String(groupID), MessageDeduplicationId: aws.String(deduplicationID), diff --git a/parser/queue/vaa_sqs.go b/parser/queue/vaa_sqs.go index 378df9e5..31b9e8d4 100644 --- a/parser/queue/vaa_sqs.go +++ b/parser/queue/vaa_sqs.go @@ -78,7 +78,7 @@ func (q *SQS) Consume(ctx context.Context) <-chan *ConsumerMessage { q.ch <- &ConsumerMessage{ Data: &body, Ack: func() { - if err := q.consumer.DeleteMessage(msg); err != nil { + if err := q.consumer.DeleteMessage(&msg); err != nil { q.logger.Error("Error deleting message from SQS", zap.Error(err)) } },