Merge pull request #157 from OasisDEX/kk/fix-sqs

FEAT - AWS SQS implementation

.env

@@ -5,3 +5,8 @@ VL_DB_HOST=localhost
 VL_DB_PORT=5432
 VL_CHAIN_HOST=
 VL_CHAIN_NAME=mainnet
+AWS_ACCESS_KEY_ID=xxx
+AWS_SECRET_ACCESS_KEY=xxx
+AWS_REGION=eu-east-1
+AWS_SQS=local
+AWS_QUEUE_URL=

README.md

@@ -127,6 +127,50 @@ event it needs to be multiplied by rate.
   ```
 - Remember to lowercase all addresses when saving them.
 
+## Sending messages to AWS SQS
+Fill in the required env variables :
+  - `AWS_ACCESS_KEY_ID=`
+  - `AWS_SECRET_ACCESS_KEY=`
+  - `AWS_REGION=`
+  - `AWS_SQS=local`
+
+To use the queue - import `aws` and use it where desired - eg `oracleTransformer.ts`.
+
+If no `AWS_SQS` is provided or it is == `local` the messages will not be sent to the queue and will be displayed in the terminal - you can use the debug console to track them.
+
+example :
+```
+ aws.sendMessage(
+      {
+        MessageAttributes: {
+          Name: {
+            DataType: 'String',
+            StringValue: `TokenPrice`,
+          },
+          Type: {
+            DataType: 'String',
+            StringValue: `OsmEvent`,
+          },
+          Value: {
+            DataType: 'String',
+            StringValue: `${price}`,
+          },
+        },
+        MessageBody: `${token}-${price}-${nextPrice}`,
+        MessageDeduplicationId: `${token}x${block.number}`,
+        MessageGroupId: token,
+        QueueUrl: process.env.AWS_QUEUE_URL,
+      },
+      function(err: any, data: any) {
+        if (err) {
+          console.log('Error', err);
+        } else {
+          console.log('Success', data.MessageId);
+        }
+      },
+    );
+```
+
 ## Troubleshooting known issues
 
 ### Getting EHOSTUNREACH on some ip when running yarn start-etl

package.json

@@ -47,6 +47,7 @@
     "@oasisdex/spock-utils": "^0.1.0",
     "@oasisdex/spock-validation": "^0.1.0",
     "abi-decoder": "^1.2.0",
+    "aws-sdk": "^2.1232.0",
     "bignumber.js": "^8.0.2",
     "consola": "^2.15.3",
     "dotenv-flow": "^3.2.0",

src/utils/awsQueue.ts

@@ -0,0 +1,24 @@
+import * as AWS from 'aws-sdk';
+
+function getAWS() {
+  if (process.env.AWS_SQS == 'production') {
+    AWS.config.update({
+      accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+      secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
+      region: process.env.AWS_REGION,
+    });
+    return new AWS.SQS({ apiVersion: '2012-11-05' });
+  } else {
+    const aws = {
+      sendMessage: (
+        params: AWS.SQS.SendMessageRequest,
+        callback?: ((err: AWS.AWSError, data: AWS.SQS.SendMessageResult) => void) | undefined,
+      ) => {
+        console.log('Queue message body ', params);
+      },
+    };
+    return aws;
+  }
+}
+const aws = getAWS();
+export { aws };