Skip to content
This repository was archived by the owner on Dec 9, 2024. It is now read-only.

Commit f673e72

Browse files
alexdebriealexdebrie
authored
Add SQSProvider (#399)
* Add SQSProvider * Update OpenAPI
1 parent 4937d70 commit f673e72

File tree

7 files changed

+1067
-6
lines changed

7 files changed

+1067
-6
lines changed

‎README.md

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ Discover and call serverless functions from anything that can reach the Event Ga
8585
following function types:
8686

8787
* FaaS functions (AWS Lambda, Google Cloud Functions, Azure Functions, OpenWhisk Actions)
88-
* Connectors (AWS Kinesis, AWS Kinesis Firehose)
88+
* Connectors (AWS Kinesis, AWS Kinesis Firehose, AWS SQS)
8989
* HTTP endpoints/Webhook (e.g. POST http://example.com/function)
9090

9191
Function Discovery stores information about functions allowing the Event Gateway to call them as a reaction to received
@@ -461,6 +461,12 @@ JSON object:
461461
* `awsAccessKeyId` - `string` - optional, AWS API key ID. By default credentials from the [environment](http://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials) are used.
462462
* `awsSecretAccessKey` - `string` - optional, AWS API access key. By default credentials from the [environment](http://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials) are used.
463463
* `awsSessionToken` - `string` - optional, AWS session token
464+
* for AWS SQS connector:
465+
* `queueUrl` - `string` - required, AWS SQS Queue URL
466+
* `region` - `string` - required, region name
467+
* `awsAccessKeyId` - `string` - optional, AWS API key ID. By default credentials from the [environment](http://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials) are used.
468+
* `awsSecretAccessKey` - `string` - optional, AWS API access key. By default credentials from the [environment](http://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html#specifying-credentials) are used.
469+
* `awsSessionToken` - `string` - optional, AWS session token
464470

465471
**Response**
466472

‎cmd/event-gateway/main.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
_ "github.com/serverless/event-gateway/providers/awsfirehose"
2727
_ "github.com/serverless/event-gateway/providers/awskinesis"
2828
_ "github.com/serverless/event-gateway/providers/awslambda"
29+
_ "github.com/serverless/event-gateway/providers/awssqs"
2930
_ "github.com/serverless/event-gateway/providers/http"
3031
)
3132

‎docs/openapi/openapi-config-api.yaml

Lines changed: 41 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -213,15 +213,19 @@ components:
213213
type: string
214214
description: "function provider"
215215
enum:
216-
- awslambda
216+
- awsfirehose
217217
- awskinesis
218+
- awslambda
219+
- awssqs
218220
- http
219221
Provider:
220222
type: object
221223
description: "function provider configuration"
222224
oneOf:
225+
- $ref: '#/components/schemas/AWSFirehose'
223226
- $ref: '#/components/schemas/AWSKinesis'
224227
- $ref: '#/components/schemas/AWSLambda'
228+
- $ref: '#/components/schemas/AWSSQS'
225229
- $ref: '#/components/schemas/HTTP'
226230
Function:
227231
type: object
@@ -265,11 +269,11 @@ components:
265269
type: array
266270
items:
267271
$ref: '#/components/schemas/Subscription'
268-
AWSLambda:
272+
AWSFirehose:
269273
type: object
270274
properties:
271-
arn:
272-
$ref: '#/components/schemas/ARN'
275+
deliveryStreamName:
276+
$ref: '#/components/schemas/DeliveryStreamName'
273277
region:
274278
$ref: '#/components/schemas/Region'
275279
awsAccessKeyId:
@@ -291,6 +295,32 @@ components:
291295
$ref: '#/components/schemas/AWSSecretAccessKey'
292296
awsSessionToken:
293297
$ref: '#/components/schemas/AWSSessionToken'
298+
AWSLambda:
299+
type: object
300+
properties:
301+
arn:
302+
$ref: '#/components/schemas/ARN'
303+
region:
304+
$ref: '#/components/schemas/Region'
305+
awsAccessKeyId:
306+
$ref: '#/components/schemas/AWSAccessKeyId'
307+
awsSecretAccessKey:
308+
$ref: '#/components/schemas/AWSSecretAccessKey'
309+
awsSessionToken:
310+
$ref: '#/components/schemas/AWSSessionToken'
311+
AWSSQS:
312+
type: object
313+
properties:
314+
queueUrl:
315+
$ref: '#/components/schemas/QueueURL'
316+
region:
317+
$ref: '#/components/schemas/Region'
318+
awsAccessKeyId:
319+
$ref: '#/components/schemas/AWSAccessKeyId'
320+
awsSecretAccessKey:
321+
$ref: '#/components/schemas/AWSSecretAccessKey'
322+
awsSessionToken:
323+
$ref: '#/components/schemas/AWSSessionToken'
294324
HTTP:
295325
type: object
296326
properties:
@@ -314,6 +344,12 @@ components:
314344
StreamName:
315345
type: string
316346
description: "AWS Kinesis stream name"
347+
DeliveryStreamName:
348+
type: string
349+
description: "AWS Firehose delivery stream name"
350+
QueueURL:
351+
type: string
352+
description: "AWS SQS Queue URL"
317353
URL:
318354
type: string
319355
format: url
@@ -456,4 +492,4 @@ components:
456492
content:
457493
application/json:
458494
schema:
459-
$ref: '#/components/schemas/Errors'
495+
$ref: '#/components/schemas/Errors'

‎providers/awssqs/awssqs.go

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
package awssqs
2+
3+
import (
4+
"encoding/json"
5+
"errors"
6+
7+
"github.com/aws/aws-sdk-go/aws"
8+
"github.com/aws/aws-sdk-go/aws/awserr"
9+
"github.com/aws/aws-sdk-go/aws/credentials"
10+
"github.com/aws/aws-sdk-go/aws/session"
11+
"github.com/aws/aws-sdk-go/service/sqs"
12+
"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
13+
"github.com/serverless/event-gateway/function"
14+
"go.uber.org/zap/zapcore"
15+
validator "gopkg.in/go-playground/validator.v9"
16+
)
17+
18+
// Type of provider.
19+
const Type = function.ProviderType("awssqs")
20+
21+
func init() {
22+
function.RegisterProvider(Type, ProviderLoader{})
23+
}
24+
25+
// AWSSQS function implementation
26+
type AWSSQS struct {
27+
Service sqsiface.SQSAPI`json:"-" validate:"-"`
28+
29+
QueueURL string `json:"queueUrl" validate:"required"`
30+
Region string `json:"region" validate:"required"`
31+
AWSAccessKeyID string `json:"awsAccessKeyId,omitempty"`
32+
AWSSecretAccessKey string `json:"awsSecretAccessKey,omitempty"`
33+
AWSSessionToken string `json:"awsSessionToken,omitempty"`
34+
}
35+
36+
// Call sends message to AWS SQS Queue
37+
func (a AWSSQS) Call(payload []byte) ([]byte, error) {
38+
body := string(payload)
39+
sendMessageOutput, err := a.Service.SendMessage(&sqs.SendMessageInput{
40+
QueueUrl: &a.QueueURL,
41+
MessageBody: &body,
42+
})
43+
if err != nil {
44+
if awserr, ok := err.(awserr.Error); ok {
45+
return nil, &function.ErrFunctionCallFailed{Original: awserr}
46+
}
47+
}
48+
49+
return []byte(*sendMessageOutput.MessageId), err
50+
}
51+
52+
// validate provider config.
53+
func (a AWSSQS) validate() error {
54+
validate := validator.New()
55+
err := validate.Struct(a)
56+
if err != nil {
57+
return err
58+
}
59+
return nil
60+
}
61+
62+
// MarshalLogObject is a part of zapcore.ObjectMarshaler interface.
63+
func (a AWSSQS) MarshalLogObject(enc zapcore.ObjectEncoder) error {
64+
enc.AddString("queueUrl", a.QueueURL)
65+
enc.AddString("region", a.Region)
66+
if a.AWSAccessKeyID != "" {
67+
enc.AddString("awsAccessKeyId", "*****")
68+
}
69+
if a.AWSSecretAccessKey != "" {
70+
enc.AddString("awsSecretAccessKey", "*****")
71+
}
72+
if a.AWSSessionToken != "" {
73+
enc.AddString("awsSessionToken", "*****")
74+
}
75+
return nil
76+
}
77+
78+
// ProviderLoader implementation
79+
type ProviderLoader struct{}
80+
81+
// Load decode JSON data as Config and return initialized Provider instance.
82+
func (p ProviderLoader) Load(data []byte) (function.Provider, error) {
83+
provider := &AWSSQS{}
84+
err := json.Unmarshal(data, provider)
85+
if err != nil {
86+
return nil, errors.New("unable to load function provider config: " + err.Error())
87+
}
88+
89+
err = provider.validate()
90+
if err != nil {
91+
return nil, errors.New("missing required fields for AWS SQS function")
92+
}
93+
94+
config := aws.NewConfig().WithRegion(provider.Region)
95+
if provider.AWSAccessKeyID != "" && provider.AWSSecretAccessKey != "" {
96+
config = config.WithCredentials(credentials.NewStaticCredentials(provider.AWSAccessKeyID, provider.AWSSecretAccessKey, provider.AWSSessionToken))
97+
}
98+
99+
awsSession, err := session.NewSession(config)
100+
if err != nil {
101+
return nil, errors.New("unable to create AWS Session: " + err.Error())
102+
}
103+
104+
provider.Service = sqs.New(awsSession)
105+
return provider, nil
106+
}

‎providers/awssqs/awssqs_test.go

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
package awssqs_test
2+
3+
import (
4+
"errors"
5+
"testing"
6+
7+
"github.com/aws/aws-sdk-go/aws"
8+
"github.com/aws/aws-sdk-go/aws/awserr"
9+
"github.com/aws/aws-sdk-go/service/sqs"
10+
"github.com/golang/mock/gomock"
11+
"github.com/serverless/event-gateway/function"
12+
"github.com/serverless/event-gateway/providers/awssqs"
13+
"github.com/serverless/event-gateway/providers/awssqs/mock"
14+
"github.com/stretchr/testify/assert"
15+
"go.uber.org/zap/zapcore"
16+
)
17+
18+
func TestLoad(t *testing.T) {
19+
for _, testCase := range loadTests {
20+
config := []byte(testCase.config)
21+
loader := awssqs.ProviderLoader{}
22+
23+
_, err := loader.Load(config)
24+
25+
assert.Equal(t, testCase.expectedError, err)
26+
}
27+
}
28+
29+
func TestCall(t *testing.T) {
30+
for _, testCase := range callTests {
31+
mockCtrl := gomock.NewController(t)
32+
defer mockCtrl.Finish()
33+
serviceMock := mock.NewMockSQSAPI(mockCtrl)
34+
serviceMock.EXPECT().SendMessage(gomock.Any()).Return(testCase.sendMessageResult, testCase.sendMessageError)
35+
36+
provider := awssqs.AWSSQS{
37+
Service: serviceMock,
38+
QueueURL: "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue",
39+
Region: "us-east-1",
40+
}
41+
42+
output, err := provider.Call([]byte("testpayload"))
43+
44+
assert.Equal(t, testCase.expectedResult, output)
45+
assert.Equal(t, testCase.expectedError, err)
46+
}
47+
}
48+
49+
func TestMarshalLogObject(t *testing.T) {
50+
for _, testCase := range logTests {
51+
enc := zapcore.NewMapObjectEncoder()
52+
53+
testCase.provider.MarshalLogObject(enc)
54+
55+
assert.Equal(t, testCase.expectedFields, enc.Fields)
56+
}
57+
}
58+
59+
var loadTests = []struct {
60+
config string
61+
expectedError error
62+
}{
63+
{
64+
`{"queueUrl": "", "region": `,
65+
errors.New("unable to load function provider config: unexpected end of JSON input"),
66+
},
67+
{
68+
`{"queueUrl": "", "region": "us-east-1"}`,
69+
errors.New("missing required fields for AWS SQS function"),
70+
},
71+
{
72+
`{"queueUrl": "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue", "region": ""}`,
73+
errors.New("missing required fields for AWS SQS function"),
74+
},
75+
}
76+
77+
var callTests = []struct {
78+
sendMessageResult *sqs.SendMessageOutput
79+
sendMessageError error
80+
expectedResult []byte
81+
expectedError error
82+
}{
83+
{
84+
&sqs.SendMessageOutput{MessageId: aws.String("testid")},
85+
nil,
86+
[]byte("testid"),
87+
nil,
88+
},
89+
{
90+
nil,
91+
awserr.New("", "", nil),
92+
[]byte(nil),
93+
&function.ErrFunctionCallFailed{Original: awserr.New("", "", nil)},
94+
},
95+
}
96+
97+
var logTests = []struct {
98+
provider function.Provider
99+
expectedFields map[string]interface{}
100+
}{
101+
{
102+
awssqs.AWSSQS{
103+
QueueURL : "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue",
104+
Region: "us-east-1",
105+
},
106+
map[string]interface{}{
107+
"queueUrl": "https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue",
108+
"region": "us-east-1",
109+
},
110+
},
111+
{
112+
awssqs.AWSSQS{
113+
AWSAccessKeyID: "id",
114+
AWSSecretAccessKey: "key",
115+
AWSSessionToken: "token",
116+
},
117+
map[string]interface{}{
118+
"queueUrl": "",
119+
"region": "",
120+
"awsAccessKeyId": "*****",
121+
"awsSecretAccessKey": "*****",
122+
"awsSessionToken": "*****",
123+
},
124+
},
125+
}

‎providers/awssqs/mock/mock.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
//go:generate mockgen -package mock -destination sqsiface.go github.com/aws/aws-sdk-go/service/sqs/sqsiface SQSAPI
2+
3+
package mock

0 commit comments

Comments
 (0)