Skip to content

Commit

Permalink
Allow for standard SQS URLs
Browse files Browse the repository at this point in the history
Both now work:

https://sqs.us-east-1.amazonaws.com/349840735605/TestTile38Queue
sqs://us-east-1:349840735605/TestTile38Queue
  • Loading branch information
tidwall committed Mar 13, 2019
1 parent ec57aae commit 5335aec
Show file tree
Hide file tree
Showing 2 changed files with 58 additions and 22 deletions.
48 changes: 28 additions & 20 deletions internal/endpoint/endpoint.go
Expand Up @@ -90,6 +90,7 @@ type Endpoint struct {
KeyFile string
}
SQS struct {
PlainURL string
QueueID string
Region string
CredPath string
Expand Down Expand Up @@ -217,7 +218,12 @@ func parseEndpoint(s string) (Endpoint, error) {
case strings.HasPrefix(s, "http:"):
endpoint.Protocol = HTTP
case strings.HasPrefix(s, "https:"):
endpoint.Protocol = HTTP
if probeSQS(s) {
endpoint.SQS.PlainURL = s
endpoint.Protocol = SQS
} else {
endpoint.Protocol = HTTP
}
case strings.HasPrefix(s, "disque:"):
endpoint.Protocol = Disque
case strings.HasPrefix(s, "grpc:"):
Expand Down Expand Up @@ -469,22 +475,28 @@ func parseEndpoint(s string) (Endpoint, error) {
// credpath - path where aws credentials are located
// credprofile - credential profile
if endpoint.Protocol == SQS {
// Parsing connection from URL string
hp := strings.Split(s, ":")
switch len(hp) {
default:
return endpoint, errors.New("invalid SQS url")
case 2:
endpoint.SQS.Region = hp[0]
endpoint.SQS.QueueID = hp[1]
}
if endpoint.SQS.PlainURL == "" {
// Parsing connection from URL string
hp := strings.Split(s, ":")
switch len(hp) {
default:
return endpoint, errors.New("invalid SQS url")
case 2:
endpoint.SQS.Region = hp[0]
endpoint.SQS.QueueID = hp[1]
}

// Parsing SQS queue name
if len(sp) > 1 {
var err error
endpoint.SQS.QueueName, err = url.QueryUnescape(sp[1])
if err != nil {
return endpoint, errors.New("invalid SQS queue name")
// Parsing SQS queue name
if len(sp) > 1 {
var err error
endpoint.SQS.QueueName, err = url.QueryUnescape(sp[1])
if err != nil {
return endpoint, errors.New("invalid SQS queue name")
}
}
// Throw error if we not provide any queue name
if endpoint.SQS.QueueName == "" {
return endpoint, errors.New("missing SQS queue name")
}
}

Expand Down Expand Up @@ -512,10 +524,6 @@ func parseEndpoint(s string) (Endpoint, error) {
}
}
}
// Throw error if we not provide any queue name
if endpoint.SQS.QueueName == "" {
return endpoint, errors.New("missing SQS queue name")
}
}

// Basic AMQP connection strings in HOOKS interface
Expand Down
32 changes: 30 additions & 2 deletions internal/endpoint/sqs.go
Expand Up @@ -3,6 +3,7 @@ package endpoint
import (
"errors"
"fmt"
"strings"
"sync"
"time"

Expand Down Expand Up @@ -31,7 +32,11 @@ type SQSConn struct {
}

func (conn *SQSConn) generateSQSURL() string {
return "https://sqs." + conn.ep.SQS.Region + "amazonaws.com/" + conn.ep.SQS.QueueID + "/" + conn.ep.SQS.QueueName
if conn.ep.SQS.PlainURL != "" {
return conn.ep.SQS.PlainURL
}
return "https://sqs." + conn.ep.SQS.Region + ".amazonaws.com/" +
conn.ep.SQS.QueueID + "/" + conn.ep.SQS.QueueName
}

// Expired returns true if the connection has expired
Expand Down Expand Up @@ -74,8 +79,14 @@ func (conn *SQSConn) Send(msg string) error {
}
creds = credentials.NewSharedCredentials(credPath, credProfile)
}
var region string
if conn.ep.SQS.Region != "" {
region = conn.ep.SQS.Region
} else {
region = sqsRegionFromPlainURL(conn.ep.SQS.PlainURL)
}
sess := session.Must(session.NewSession(&aws.Config{
Region: aws.String(conn.ep.SQS.Region),
Region: &region,
Credentials: creds,
MaxRetries: aws.Int(5),
}))
Expand Down Expand Up @@ -114,3 +125,20 @@ func newSQSConn(ep Endpoint) *SQSConn {
t: time.Now(),
}
}

func probeSQS(s string) bool {
// https://sqs.eu-central-1.amazonaws.com/123456789/myqueue
return strings.HasPrefix(s, "https://sqs.") &&
strings.Contains(s, ".amazonaws.com/")
}

func sqsRegionFromPlainURL(s string) string {
parts := strings.Split(s, "https://sqs.")
if len(parts) > 1 {
parts = strings.Split(parts[1], ".amazonaws.com/")
if len(parts) > 1 {
return parts[0]
}
}
return ""
}

0 comments on commit 5335aec

Please sign in to comment.