Configure Queue

Ply includes a built-in background job system for running asynchronous tasks with support for in-memory and persistent file storage.

Initialization

Initialize the queue system in your main.go:

app := ply.New().WithQueue()
controller.Quehandler(app.EngineCtx)

You can also chain this with other modules:

app := ply.New().WithCache().WithPubsub().WithQueue()

How It Works

• Jobs are pushed with Enqueue(topic, data) and processed later.
• Each queue is identified by a topic string.
• Workers process jobs from either memory, storage, or both.

Example usage

Enqueue to memory (fast, volatile)

if err := c.MustQueue().Memory.Enqueue("user:welcome-email", payload); err != nil {
    c.JSON(500, map[string]string{"error": "Queue error"})
    return
}

Enqueue to storage (durable, file-based)

if err := c.MustQueue().Storage.Enqueue("user:welcome-email", payload); err != nil {
    c.JSON(500, map[string]string{"error": "Queue error"})
    return
}

Default behavior

When using:

c.MustQueue().EnqueueAll("user:welcome-email", payload)

Ply will enqueue the job to both memory and storage (combined mode).

Registering workers

Ply allows flexible worker registration depending on delivery mode:

// Memory worker only
c.Queue.Memory.RegisterWorker("user:welcome-email", SendWelcomeEmailWorker)

// Storage worker only
c.Queue.Storage.RegisterWorker("user:welcome-email", SendWelcomeEmailWorker)

// Both memory + storage (default)
c.Queue.RegisterWorkerAll("user:welcome-email", SendWelcomeEmailWorker)

Worker example:

func SendWelcomeEmailWorker(data []byte) error {
	var input CreateUserInput
	if err := json.Unmarshal(data, &input); err != nil {
		return fmt.Errorf("invalid payload: %w", err)
	}
	fmt.Printf("Queue worker: %s <%s>\n", input.Name, input.Email)
	return nil
}

func Quehandler(ctx *core.EngineContext) {
	if ctx.Queue == nil {
		panic("Queue engine not initialized")
	}
	ctx.Queue.RegisterWorkerAll("user:welcome-email", SendWelcomeEmailWorker)
}

Configuration modes

Ply Queue supports three modes:

• Memory Only: Temporary, best for speed in dev/test.
• File Storage Only: Ensures durability across restarts.
• Combined (default): uses both for speed and durability.

Behind the scenes

Ply Queue uses a dual-layer system:

• Memory: Fast, buffered in-RAM processing
• Storage: Durable JSON logs for crash recovery

Jobs can be enqueued to either or both layers. Automatic cleanup runs every 10 minutes, removing jobs older than 24 hours.

Auto cleanup

Ply automatically cleans up old queue files:

• Interval: every 10 * time.Minute
• Retention: deletes jobs older than 24 * time.Hour

No manual intervention needed.

Use Cases

• Email delivery (welcome, verification)
• Background processing (thumbnails, reports)
• Asynchronous workflows