wip

.github/workflows/fifoqueue-pull.yaml

@@ -0,0 +1,29 @@
+name: fifoqueue-pull
+on:
+  pull_request:
+    paths:
+      - fifoqueue/**
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          sparse-checkout: fifoqueue
+      - name: Setup Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: 18.x
+          registry-url: https://registry.npmjs.org
+      - name: Install winglang
+        run: npm i -g winglang
+      - name: Install dependencies
+        run: npm install --include=dev
+        working-directory: fifoqueue
+      - name: Test
+        run: wing test
+        working-directory: fifoqueue
+      - name: Pack
+        run: wing pack
+        working-directory: fifoqueue

.github/workflows/fifoqueue-release.yaml

@@ -0,0 +1,37 @@
+name: fifoqueue-release
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - fifoqueue/**
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          sparse-checkout: fifoqueue
+      - name: Setup Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: 18.x
+          registry-url: https://registry.npmjs.org
+      - name: Install winglang
+        run: npm i -g winglang
+      - name: Install dependencies
+        run: npm install --include=dev
+        working-directory: fifoqueue
+      - name: Test
+        run: wing test
+        working-directory: fifoqueue
+      - name: Pack
+        run: wing pack
+        working-directory: fifoqueue
+      - name: Publish
+        run: npm publish --access=public --registry https://registry.npmjs.org --tag
+          latest *.tgz
+        working-directory: fifoqueue
+        env:
+          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

fifoqueue/.gitignore

@@ -0,0 +1,2 @@
+target/
+node_modules/

fifoqueue/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Wing
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

fifoqueue/README.md

@@ -0,0 +1,33 @@
+# fifoqueue
+
+A wing library to work with FIFO (first-in first-out) Queues.
+
+## Prerequisites
+
+* [winglang](https://winglang.io).
+
+## Installation
+
+`sh
+npm i @winglibs/fifoqueue
+`
+
+## Usage
+
+`js
+bring fifoqueue;
+
+let queue = new fifoqueue.FifoQueue();
+
+queue.setConsumer(inflight (message: str) => {
+  log("recieved message {message}");
+});
+
+test "will push to queue" {
+  queue.push("a new message");
+}
+`
+
+## License
+
+This library is licensed under the [MIT License](./LICENSE).

fifoqueue/api.w

@@ -0,0 +1,14 @@
+bring cloud;
+
+pub struct FifoQueueProps extends cloud.QueueProps{}
+
+pub struct PushOptions {
+  groupId: str;
+}
+
+pub struct SetConsumerOptions extends cloud.QueueSetConsumerOptions {}
+
+pub interface IFifoQueue {
+  setConsumer(fn: inflight (str): void, options: SetConsumerOptions?);
+  inflight push(message: str, options: PushOptions);
+}

fifoqueue/aws.ts

@@ -0,0 +1,13 @@
+import {
+  SQSClient,
+  SendMessageCommand,
+} from "@aws-sdk/client-sqs";
+
+export const _push = async (queueUrl: string, message: string, groupId: string) => {
+  const client = new SQSClient();
+  client.send(new SendMessageCommand({
+    QueueUrl: queueUrl,
+    MessageBody: message,
+    MessageGroupId: groupId,
+  }))
+};

fifoqueue/fifo-queue.aws.w

@@ -0,0 +1,84 @@
+bring "cdktf" as cdktf;
+bring "@cdktf/provider-aws" as aws;
+bring cloud;
+bring aws as awsUtil;
+bring "./api.w" as api;
+
+struct Record {
+  body: str;
+}
+
+struct SqsEvent {
+  Records: Array<Record>;
+}
+
+pub class FifoQueue_aws impl api.IFifoQueue {
+  queue: aws.sqsQueue.SqsQueue;
+  url: str;
+  arn: str;
+  new(props: api.FifoQueueProps?) {
+    this.queue = new aws.sqsQueue.SqsQueue(
+      visibilityTimeoutSeconds: props?.timeout?.seconds
+        ?? duration.fromSeconds(120).seconds,
+      messageRetentionSeconds: props?.retentionPeriod?.seconds
+        ?? duration.fromHours(1).seconds,
+      fifoQueue: true,
+      contentBasedDeduplication: true,
+    );
+    this.url = this.queue.url;
+    this.arn = this.queue.arn;
+  }
+
+  pub setConsumer(fn: inflight (str) : void, options: api.SetConsumerOptions?) {
+    let lambdaFn = new cloud.Function(inflight (event: str): void => {
+      let json: Json = unsafeCast(event);
+      let sqsEvent = SqsEvent.fromJson(event);
+      for message in sqsEvent.Records {
+        fn(message.body);
+      }
+    },
+      env: options?.env, 
+      logRetentionDays: options?.logRetentionDays, 
+      memory: options?.memory, 
+      timeout: options?.timeout
+    );
+
+    let lambda = awsUtil.Function.from(lambdaFn);
+    lambda?.addPolicyStatements({
+      actions: [
+        "sqs:ReceiveMessage",
+        "sqs:ChangeMessageVisibility",
+        "sqs:GetQueueUrl",
+        "sqs:DeleteMessage",
+        "sqs:GetQueueAttributes",
+      ],
+      resources: [this.arn],
+    });
+
+    new aws.lambdaEventSourceMapping.LambdaEventSourceMapping(
+      functionName: "{lambda?.functionName}",
+      eventSourceArn: this.arn,
+      batchSize: options?.batchSize ?? 1
+    );
+  }
+
+  pub onLift(host: std.IInflightHost, ops: Array<str>) {
+    if let lambda = awsUtil.Function.from(host) {
+      if ops.contains("push") {
+        lambda.addPolicyStatements({
+          actions: ["sqs:SendMessage"],
+          effect: awsUtil.Effect.ALLOW,
+          resources: [
+            this.arn
+          ]
+        });
+      }
+    }
+  }
+
+  pub inflight push(message: str, options: api.PushOptions) {
+    FifoQueue_aws._push(this.url, message, options.groupId);
+  }
+
+  extern "./aws.ts" static inflight _push(queueUrl: str, message: str, groupId: str); 
+}

fifoqueue/fifo-queue.sim.w

@@ -0,0 +1,47 @@
+bring expect;
+bring util;
+bring cloud;
+bring "./api.w" as api;
+
+struct FifoQueueMessage  {
+  groupId: str;
+  message: str;
+}
+
+pub class FifoQueue_sim impl api.IFifoQueue { 
+  queue: cloud.Queue;
+  counter: cloud.Counter;
+
+  new(){
+    this.queue = new cloud.Queue();
+    this.counter = new cloud.Counter();
+  }
+
+  pub setConsumer(fn: inflight (str): void, options: api.SetConsumerOptions?) {
+    let counter = this.counter;
+    this.queue.setConsumer(inflight (m: str) => {
+      let j = FifoQueueMessage.parseJson(m);
+      util.waitUntil(inflight () => {
+        let value = counter.peek(j.groupId);
+        if value == 0 {
+          let acquired = counter.inc(1, j.groupId);
+          if acquired == 0 {
+            return true;
+          } else {
+            counter.dec(1, j.groupId);
+            return false;
+          }
+        }
+        return false;
+      }, timeout: 30m);
+
+      fn(j.message);
+      counter.dec(1, j.groupId);
+    });
+
+  }
+
+  inflight pub push(message: str, options: api.PushOptions) {
+    this.queue.push(Json.stringify({ groupId: options.groupId, message: message }));
+  }
+}

fifoqueue/fifo-queue.test.w

@@ -0,0 +1,23 @@
+bring expect;
+bring util;
+bring cloud;
+bring "./fifo-queue.w" as fifo_queue;
+
+let queue = new fifo_queue.FifoQueue();
+let counter = new cloud.Counter();
+
+queue.setConsumer(inflight (message: str) => {
+  log("start {message}");
+  util.sleep(10s);
+  counter.inc();
+  log("end {message}");
+});
+
+test "sequentially consume messages with the same group id " {
+  queue.push("msg1", groupId: "123");
+  queue.push("msg2", groupId: "123");
+  util.sleep(15s);
+  expect.equal(counter.peek(), 1);
+  util.sleep(15s);
+  expect.equal(counter.peek(), 2);
+}

fifoqueue/fifo-queue.w

@@ -0,0 +1,27 @@
+bring util;
+bring "./api.w" as api;
+bring "./fifo-queue.sim.w" as sim;
+bring "./fifo-queue.aws.w" as aws;
+
+pub class FifoQueue impl api.IFifoQueue {
+  inner: api.IFifoQueue;
+  new(props: api.FifoQueueProps?) {
+    let target = util.env("WING_TARGET");
+    if target == "sim" {
+      this.inner = new sim.FifoQueue_sim() as "sim";
+    } elif target == "tf-aws" {
+      this.inner = new aws.FifoQueue_aws(props) as "tf-aws";
+    } else {
+      throw "Unsupported target {target}";
+    }
+  }
+
+  pub setConsumer(fn: inflight (str): void, options: api.SetConsumerOptions?) {
+    this.inner.setConsumer(fn, options);
+
+  }
+
+  pub inflight push(message: str, options: api.PushOptions) {
+    this.inner.push(message, options);
+  }
+}