deno-pgmq

import { Pgmq } from "https://deno.land/x/pgmq@v0.2.1/mod.ts";

console.log("Connecting to Postgres...");
// Specify the connection parameters manually
let pgmq: Pgmq;
try {
  pgmq = await Pgmq.new({
    dsn:
      "postgresql://postgres:postgres@127.0.0.1:54322/postgres?sslmode=require",
    // Optional parameters
    caFilePaths: ["./certs/prod-ca-2021.crt"],
  });
} catch (err) {
  console.error("Failed to connect to Postgres", err);
  Deno.exit(1);
}

// You can also use environment variables to set the connection parameters
// $ export DATABASE_URL='postgresql://postgres:postgres@localhost:54322/postgres'
// $ export LAZY=true
// $ export MAX_POOL_SIZE=20
// const pgmq = await Pgmq.new()

const qName = "my_queue";
console.log(`Creating queue ${qName}...`);

await pgmq.queue.create(qName).catch((err) => {
  console.error("Failed to create queue", err);
  Deno.exit(1);
});

interface Msg {
  id: number;
  name: string;
}

const msg: Msg = { id: 1, name: "testMsg" };
console.log("Sending message...");
const msgId = await pgmq.msg.send(qName, msg).catch((err) => {
  console.error("Failed to send message", err);
  Deno.exit(1);
});

const vt = 30;
const receivedMsg = await pgmq.msg.read<Msg>(qName, vt).catch((err) => {
  console.error("No messages in the queue", err);
  Deno.exit(1);
});

console.log("Received message...");
if (receivedMsg) {
  console.dir(receivedMsg.message, { depth: null });
} else {
  console.log("No message received.");
}

console.log("Archiving message...");
await pgmq.msg.archive(qName, msgId).catch((err) => {
  console.error("Failed to archive message", err);
  Deno.exit(1);
});