Merge pull request #24 from horike37/change-behavior-kinesis

feat: change default behavior of kinesis proxy

Author: horike37

README.md

@@ -33,9 +33,36 @@ Sample syntax for Kinesis proxy in `serverless.yml`.
 ```yaml
 custom:
   apiGatewayServiceProxies:
+    - kinesis: # partitionkey is set apigateway requestid by default
+        path: /kinesis
+        method: post
+        streamName: { Ref: 'YourStream' }
+        cors: true
+    - kinesis:
+        path: /kinesis
+        method: post
+        partitionKey: 'hardcordedkey' # use static partitionkey
+        streamName: { Ref: 'YourStream' }
+        cors: true
+    - kinesis:
+        path: /kinesis/{myKey} # use path parameter
+        method: post
+        partitionKey:
+          pathParam: myKey
+        streamName: { Ref: 'YourStream' }
+        cors: true
+    - kinesis:
+        path: /kinesis
+        method: post
+        partitionKey:
+          bodyParam: data.myKey # use body parameter
+        streamName: { Ref: 'YourStream' }
+        cors: true
     - kinesis:
         path: /kinesis
         method: post
+        partitionKey:
+          queryStringParam: myKey # use query string param
         streamName: { Ref: 'YourStream' }
         cors: true
 
@@ -50,7 +77,7 @@ resources:
 Sample request after deploying.
 
 ```bash
-curl -X POST https://xxxxxxx.execute-api.us-east-1.amazonaws.com/dev/kinesis -d '{"Data": "some data","PartitionKey": "some key"}'  -H 'Content-Type:application/json'
+curl -X POST https://xxxxxxx.execute-api.us-east-1.amazonaws.com/dev/kinesis -d '{"message": "some data"}'  -H 'Content-Type:application/json'
 ```
 
 ### SQS
@@ -78,6 +105,28 @@ Sample request after deploying.
 curl -X POST https://xxxxxx.execute-api.us-east-1.amazonaws.com/dev/sqs -d '{"message": "testtest"}' -H 'Content-Type:application/json'
 ```
 
+#### Customizing request parameters
+
+If you'd like to pass additional data to the integration request, you can do so by including your custom [API Gateway request parameters](https://docs.aws.amazon.com/apigateway/latest/developerguide/request-response-data-mappings.html) in `serverless.yml` like so:
+
+```yml
+custom:
+  apiGatewayServiceProxies:
+    - sqs:
+        path: /queue
+        method: post
+        queueName: !GetAtt MyQueue.QueueName
+        cors: true
+
+        requestParameters:
+          'integration.request.querystring.MessageAttribute.1.Name': "'cognitoIdentityId'"
+          'integration.request.querystring.MessageAttribute.1.Value.StringValue': 'context.identity.cognitoIdentityId'
+          'integration.request.querystring.MessageAttribute.1.Value.DataType': "'String'"
+          'integration.request.querystring.MessageAttribute.2.Name': "'cognitoAuthenticationProvider'"
+          'integration.request.querystring.MessageAttribute.2.Value.StringValue': 'context.identity.cognitoAuthenticationProvider'
+          'integration.request.querystring.MessageAttribute.2.Value.DataType': "'String'"
+```
+
 ### S3
 
 Sample syntax for S3 proxy in `serverless.yml`.
@@ -231,11 +280,7 @@ resources:
 
 Source: [AWS::ApiGateway::Method docs](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-apigateway-method.html#cfn-apigateway-method-authorizationtype)
 
-## Specific features
-
-### Kinesis
-
-#### Customizing request body mapping templates
+### Customizing request body mapping templates
 
 If you'd like to add content types or customize the default templates, you can do so by including your custom [API Gateway request mapping template](https://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html) in `serverless.yml` like so:
 
@@ -263,27 +308,3 @@ custom:
 ```
 
 Source: [How to connect SNS to Kinesis for cross-account delivery via API Gateway](https://theburningmonk.com/2019/07/how-to-connect-sns-to-kinesis-for-cross-account-delivery-via-api-gateway/)
-
-### SQS
-
-#### Customizing request parameters
-
-If you'd like to pass additional data to the integration request, you can do so by including your custom [API Gateway request parameters](https://docs.aws.amazon.com/apigateway/latest/developerguide/request-response-data-mappings.html) in `serverless.yml` like so:
-
-```yml
-custom:
-  apiGatewayServiceProxies:
-    - sqs:
-        path: /queue
-        method: post
-        queueName: !GetAtt MyQueue.QueueName
-        cors: true
-
-        requestParameters:
-          'integration.request.querystring.MessageAttribute.1.Name': "'cognitoIdentityId'"
-          'integration.request.querystring.MessageAttribute.1.Value.StringValue': 'context.identity.cognitoIdentityId'
-          'integration.request.querystring.MessageAttribute.1.Value.DataType': "'String'"
-          'integration.request.querystring.MessageAttribute.2.Name': "'cognitoAuthenticationProvider'"
-          'integration.request.querystring.MessageAttribute.2.Value.StringValue': 'context.identity.cognitoAuthenticationProvider'
-          'integration.request.querystring.MessageAttribute.2.Value.DataType': "'String'"
-```

__tests__/integration/kinesis/multiple-integrations/service/serverless.yml

@@ -14,32 +14,39 @@ custom:
     - kinesis:
         path: /kinesis1
         method: post
-        streamName: { Ref: 'YourStream1' }
+        streamName: { Ref: 'YourStream' }
         cors: true
-
     - kinesis:
         path: /kinesis2
         method: post
-        streamName: { Ref: 'YourStream2' }
+        partitionKey: 'hardcordedkey' # use static partitionkey
+        streamName: { Ref: 'YourStream' }
+        cors: true
+    - kinesis:
+        path: /kinesis3/{myKey} # use path parameter
+        method: post
+        partitionKey:
+          pathParam: myKey
+        streamName: { Ref: 'YourStream' }
+        cors: true
+    - kinesis:
+        path: /kinesis4
+        method: post
+        partitionKey:
+          bodyParam: data.myKey # use body parameter
+        streamName: { Ref: 'YourStream' }
         cors: true
-
     - kinesis:
-        path: /kinesis3
+        path: /kinesis5
         method: post
-        streamName: { Ref: 'YourStream3' }
+        partitionKey:
+          queryStringParam: myKey # use query string param
+        streamName: { Ref: 'YourStream' }
         cors: true
 
 resources:
   Resources:
-    YourStream1:
-      Type: AWS::Kinesis::Stream
-      Properties:
-        ShardCount: 1
-    YourStream2:
-      Type: AWS::Kinesis::Stream
-      Properties:
-        ShardCount: 1
-    YourStream3:
+    YourStream:
       Type: AWS::Kinesis::Stream
       Properties:
         ShardCount: 1

__tests__/integration/kinesis/multiple-integrations/tests.js

@@ -19,22 +19,81 @@ describe('Multiple Kinesis Proxy Integrations Test', () => {
     removeService(stage, config)
   })
 
-  it('should get correct response from multiple kinesis proxy endpoints', async () => {
-    const streams = ['kinesis1', 'kinesis2', 'kinesis3']
-
-    for (const stream of streams) {
-      const testEndpoint = `${endpoint}/${stream}`
-
-      const response = await fetch(testEndpoint, {
-        method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ Data: `data for stream ${stream}`, PartitionKey: 'some key' })
-      })
-      expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
-      expect(response.status).to.be.equal(200)
-      const body = await response.json()
-      expect(body).to.have.own.property('ShardId')
-      expect(body).to.have.own.property('SequenceNumber')
-    }
+  it('should get correct response from kinesis proxy endpoints with default partitionkey', async () => {
+    const stream = 'kinesis1'
+    const testEndpoint = `${endpoint}/${stream}`
+    const response = await fetch(testEndpoint, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ message: `data for stream ${stream}` })
+    })
+    expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
+    expect(response.status).to.be.equal(200)
+    const body = await response.json()
+    expect(body).to.have.own.property('ShardId')
+    expect(body).to.have.own.property('SequenceNumber')
+  })
+
+  it('should get correct response from kinesis proxy endpoints with hardcorded partitionkey', async () => {
+    const stream = 'kinesis2'
+    const testEndpoint = `${endpoint}/${stream}`
+    const response = await fetch(testEndpoint, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ message: `data for stream ${stream}` })
+    })
+    expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
+    expect(response.status).to.be.equal(200)
+    const body = await response.json()
+    expect(body).to.have.own.property('ShardId')
+    expect(body).to.have.own.property('SequenceNumber')
+  })
+
+  it('should get correct response from kinesis proxy endpoints with pathparam partitionkey', async () => {
+    const stream = 'kinesis3'
+    const partitionkey = 'partitionkey'
+    const testEndpoint = `${endpoint}/${stream}/${partitionkey}`
+    const response = await fetch(testEndpoint, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ message: `data for stream ${stream}` })
+    })
+    expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
+    expect(response.status).to.be.equal(200)
+    const body = await response.json()
+    expect(body).to.have.own.property('ShardId')
+    expect(body).to.have.own.property('SequenceNumber')
+  })
+
+  it('should get correct response from kinesis proxy endpoints with bodyparam partitionkey', async () => {
+    const stream = 'kinesis4'
+    const partitionkey = 'partitionkey'
+    const testEndpoint = `${endpoint}/${stream}`
+    const response = await fetch(testEndpoint, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ message: `data for stream ${stream}`, data: { myKey: partitionkey } })
+    })
+    expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
+    expect(response.status).to.be.equal(200)
+    const body = await response.json()
+    expect(body).to.have.own.property('ShardId')
+    expect(body).to.have.own.property('SequenceNumber')
+  })
+
+  it('should get correct response from kinesis proxy endpoints with queryStringParam partitionkey', async () => {
+    const stream = 'kinesis5'
+    const partitionkey = 'partitionkey'
+    const testEndpoint = `${endpoint}/${stream}?myKey=${partitionkey}`
+    const response = await fetch(testEndpoint, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ message: `data for stream ${stream}`, data: { mykey: partitionkey } })
+    })
+    expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
+    expect(response.status).to.be.equal(200)
+    const body = await response.json()
+    expect(body).to.have.own.property('ShardId')
+    expect(body).to.have.own.property('SequenceNumber')
   })
 })

__tests__/integration/kinesis/single-integration/tests.js

@@ -25,7 +25,7 @@ describe('Single Kinesis Proxy Integration Test', () => {
     const response = await fetch(testEndpoint, {
       method: 'POST',
       headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify({ Data: 'some data', PartitionKey: 'some key' })
+      body: JSON.stringify({ message: 'some data' })
     })
     expect(response.headers.get('access-control-allow-origin')).to.deep.equal('*')
     expect(response.status).to.be.equal(200)

lib/package/kinesis/compileMethodsToKinesis.js

@@ -126,6 +126,26 @@ module.exports = {
     }
   },
 
+  getKinesisObjectRequestParameter(http) {
+    if (!_.has(http, 'partitionKey')) {
+      return '$context.requestId'
+    }
+
+    if (http.partitionKey.pathParam) {
+      return `$input.params().path.${http.partitionKey.pathParam}`
+    }
+
+    if (http.partitionKey.queryStringParam) {
+      return `$input.params().querystring.${http.partitionKey.queryStringParam}`
+    }
+
+    if (http.partitionKey.bodyParam) {
+      return `$util.parseJson($input.body).${http.partitionKey.bodyParam}`
+    }
+
+    return `${http.partitionKey}`
+  },
+
   buildDefaultKinesisRequestTemplate(http) {
     let streamName
     if (typeof http.streamName == 'object') {
@@ -134,6 +154,8 @@ module.exports = {
       streamName = `"${http.streamName}"`
     }
 
+    const objectRequestParam = this.getKinesisObjectRequestParameter(http)
+
     return {
       'Fn::Join': [
         '',
@@ -142,8 +164,8 @@ module.exports = {
           '"StreamName": "',
           streamName,
           '",',
-          '"Data": "$util.base64Encode($input.json(\'$.Data\'))",',
-          '"PartitionKey": "$input.path(\'$.PartitionKey\')"',
+          '"Data": "$util.base64Encode($input.body)",',
+          `"PartitionKey": "${objectRequestParam}"`,
           '}'
         ]
       ]