Build a Distributed Casbin Watcher with Workerman Redis in PHP

Overview

Casbin is a powerful, high‑performance open‑source access‑control framework that supports many languages such as Go, Java, Node.js, Python, PHP, .NET, C++, and Rust.

Observer

When using a distributed message system like etcd, multiple Casbin Enforcer instances need to stay consistent. A Watcher enables several Enforcer processes to share policy updates, allowing high‑throughput permission checks across instances.

Principle

In memory‑resident frameworks (e.g., Swoole, Workerman, ReactPHP) run in multi‑process mode, where each process has isolated memory. Therefore, a mechanism is required to propagate policy changes without causing memory leaks or garbage‑collection issues.

Scenario

When the policy inside an Enforcer changes, the Watcher pushes a message to a message queue (MQ). Enforcer instances listening to that queue receive the notification and automatically refresh their in‑memory policy.

Note: In a PHP‑FPM environment a Watcher is unnecessary because each request creates a fresh Enforcer instance in an isolated FPM process.

Implementation

The solution uses Workerman's asynchronous Redis client workerman/redis and its publish‑subscribe pattern.

Publish‑Subscribe Example

Interface Implementation

First, implement the official Watcher interface, which defines three methods:

/**
 * @author Tinywan (ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

interface Watcher {
    /**
     * Sets the callback that will be called when the policy in the DB is changed by other instances.
     * @param Closure $func
     */
    public function setUpdateCallback(Closure $func): void;

    /**
     * Called after the policy in the DB is changed (e.g., savePolicy, addPolicy, removePolicy).
     */
    public function update(): void;

    /**
     * Stops and releases the watcher; the callback will no longer be invoked.
     */
    public function close(): void;
}

Function Details setUpdateCallback(Closure $func) – registers an anonymous function that will be executed when another instance updates the policy. update() – publishes a message (e.g., to the /casbin channel) so other Enforcer instances can reload their policies. close() – releases resources and stops the watcher.

Concrete Implementation Class

/**
 * @author Tinywan (ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

use Casbin\Persist\Watcher;
use Closure;
use Workerman\Redis\Client;

class RedisWatcher implements Watcher {
    private Closure $callback;
    private $pubRedis;
    private $subRedis;
    private $channel;

    public function __construct(array $config) {
        $this->pubRedis = $this->createRedisClient($config);
        $this->subRedis = $this->createRedisClient($config);
        $this->channel = $config['channel'] ?? '/casbin';
        $this->subRedis->subscribe([$this->channel], function ($channel, $message) {
            if ($this->callback) {
                call_user_func($this->callback);
            }
        });
    }

    public function setUpdateCallback(Closure $func): void {
        $this->callback = $func;
    }

    public function update(): void {
        $this->pubRedis->publish($this->channel, 'casbin rules updated');
    }

    public function close(): void {
        $this->pubRedis->close();
        $this->subRedis->close();
    }

    private function createRedisClient(array $config): Client {
        $config['host'] = $config['host'] ?? '127.0.0.1';
        $config['port'] = $config['port'] ?? 6379;
        $config['password'] = $config['password'] ?? '';
        $config['database'] = $config['database'] ?? 0;
        $redis = new Client('redis://' . $config['host'] . ':' . $config['port']);
        $redis->auth($config['password'] ?? '');
        return $redis;
    }
}

Usage

/**
 * @author Tinywan (ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

use Tinywan\Casbin\Watcher\RedisWatcher;
use Casbin\Enforcer;

$enforcer = new Enforcer('path/to/model.conf', 'path/to/policy.csv');

$watcher = new RedisWatcher([
    'host' => '127.0.0.1',
    'password' => '123456',
    'port' => 6379,
    'database' => 0,
]);

// Set the watcher for the enforcer
$enforcer->setWatcher($watcher);

// Register a callback to reload the policy when an update is received
$watcher->setUpdateCallback(function () use ($enforcer) {
    $enforcer->loadPolicy();
});

This example demonstrates how to achieve real‑time policy synchronization across multiple PHP processes by leveraging Workerman's asynchronous Redis client and Casbin's Watcher interface.