dart_pgmq 2.0.0

A dart client for Postgres Message Queue (PGMQ).

Dart PGMQ

A dart client for Postgres Message Queue (PGMQ).

Setup

# Start a Postgres instance
docker run -d --name postgres -e POSTGRES_PASSWORD=postgres -p 5432:5432 quay.io/tembo/pg16-pgmq:latest

copied to clipboard

# Connect to Postgres
psql postgres://postgres:postgres@0.0.0.0:5432/postgres

copied to clipboard

-- create the pgmq schema
CREATE SCHEMA pgmq;

copied to clipboard

-- create the extension in the "pgmq" schema
CREATE EXTENSION pgmq;

copied to clipboard

Usage

• With Postgresql

// Create a database connection
final databaseParam = DatabaseConnection(
      host: 'localhost',
      database: 'postgres',
      password: 'postgres',
      username: 'postgres',
      ssl: false,
      port: 5432);

copied to clipboard

// Create a pgmq connection with DatabaseConnection param
final pgmq = await Pgmq.createConnection(param: databaseParam);

copied to clipboard

// Create a queue [queueName]
final queue = await pgmq.createQueue(queueName: queueName);

copied to clipboard

• With Prisma

// Create a pgmq connection with PrismaClient
final pgmq = Pgmq.createConnectionUsingPrisma(prismaClient: PrismaClient());

copied to clipboard

// Create a queue [queueName]
final queue = await pgmq.createQueue(queueName: queueName);

copied to clipboard

Then

// Send a message
queue.send({"foo": "bar"});

copied to clipboard

// Read a message with visibilityTimeOut
queue.read(visibilityTimeOut: vt);

copied to clipboard

// Archive a message [messageID]
queue.archive(messageID);

copied to clipboard

// Delete a message [messageID]
queue.delete(messageID);

copied to clipboard

// Pull messages from queue with a specified polling duration
queue.pull(duration: duration);

copied to clipboard

// Pausable pull
final (pausableTimer, stream) = queue.pausablePull(duration: duration);

// Start the pausable pull
pausableTimer.start();

// Pause the pulling
pausableTimer.pause();

copied to clipboard

// Read a message from queue and delete it upon read
queue.pop();

copied to clipboard

// Purge all messages from queue
queue.purgeQueue();

copied to clipboard

// Drops the queue
queue.dropQueue();

copied to clipboard

Supported Functionalities

• ❌ Sending Messages
  • ✅ send
  • ❌ send_batch
• ❌ Reading Messages
  • ✅ read
  • ❌ read_with_poll
  • ✅ pop
• ❌ Deleting/Archiving Messages
  • ✅ delete (single)
  • ❌ delete (batch)
  • ✅ purge_queue
  • ✅ archive (single)
  • ❌ archive (batch)
• ❌ Queue Management
  • ✅ create
  • ❌ create_partitioned
  • ❌ create_unlogged
  • ❌ detach_archive
  • ✅ drop_queue
• ❌ Utilities
  • ✅ set_vt
  • ❌ list_queues
  • ❌ metrics
  • ❌ metrics_all

Customs Features

• ✅ Pausable Queue