PHP自定义流过滤器（Stream Filter）：实现用户空间协议的上下文管理与数据处理 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

PHP 自定义流过滤器：上下文管理与用户空间协议实现

大家好，今天我们来深入探讨 PHP 中自定义流过滤器这一强大的特性。流过滤器允许我们在读取或写入数据流时，动态地修改和处理数据，为实现用户空间协议、数据转换、加密解密等功能提供了极大的灵活性。

一、流过滤器的基本概念

在 PHP 中，流代表了数据的来源或去向，例如文件、网络连接、内存数据等。流过滤器则是在流的读取或写入过程中，插入的一系列处理步骤，用于修改、转换或验证数据。

PHP 内置了一些流过滤器，例如 string.rot13（ROT13 编码）、convert.base64-encode（Base64 编码）等。但更重要的是，PHP 允许我们自定义流过滤器，从而实现更复杂、更定制化的数据处理逻辑。

二、自定义流过滤器的组成

一个自定义流过滤器通常由以下几个部分组成：

过滤器类： 继承自 php_user_filter 类，包含过滤器逻辑的核心实现。
filter() 方法： 这是过滤器类中最重要的方法，负责实际的数据处理。它接收输入数据、修改数据，并将处理后的数据传递给下一个过滤器或流。
onCreate() 方法： 在过滤器实例创建时调用，可以用于初始化资源、设置参数等。
onClose() 方法： 在过滤器实例销毁时调用，用于释放资源、清理状态等。
stream_filter_register() 函数： 将过滤器类注册到 PHP 中，使其可以通过名称进行调用。
stream_filter_append() 或 stream_filter_prepend() 函数： 将注册的过滤器附加到流上。

三、php_user_filter 类的详细介绍

php_user_filter 是所有自定义流过滤器的基类。它提供了一些关键的属性和方法，供子类使用。

属性/方法	说明
`$stream`	流资源，代表与过滤器关联的流。
`$params`	在附加过滤器时传递给 `stream_filter_append()` 或 `stream_filter_prepend()` 的参数。
`filter($in, $out, &$consumed, $closing)`	核心方法。处理数据块。`$in` 是包含输入数据的 bucket brigade，`$out` 是用于存储输出数据的 bucket brigade，`$consumed` 必须增加以表明 `$in` 中有多少字节被处理，`$closing` 指示流是否正在关闭。返回值必须是 `PSFS_PASS_ON`、`PSFS_FEED_ME` 或 `PSFS_ERR_FATAL`。
`onCreate()`	在过滤器创建时调用。返回 `true` 表示成功，`false` 表示失败。
`onClose()`	在过滤器关闭时调用。

filter() 方法的返回值：

PSFS_PASS_ON: 数据已处理，传递给下一个过滤器或流。
PSFS_FEED_ME: 过滤器需要更多的数据才能完成处理。
PSFS_ERR_FATAL: 发生致命错误，流操作终止。

四、使用 Bucket Brigades 处理数据

$in 和 $out 参数是 Bucket Brigade 类型的对象。Bucket Brigade 是一系列 Bucket 的集合，每个 Bucket 包含一部分数据。这种设计允许过滤器处理任意大小的数据块，而无需一次性加载所有数据到内存中。

要处理 Bucket Brigade 中的数据，我们需要遍历每个 Bucket，并从 Bucket 中读取数据。以下是一些常用的 Bucket Brigade 和 Bucket 操作：

stream_bucket_make_writeable(): 创建一个可写的 Bucket，用于存储处理后的数据。
stream_bucket_append(): 将 Bucket 添加到 Bucket Brigade 中。
stream_bucket_prepend(): 将 Bucket 添加到 Bucket Brigade 的开头。
$bucket->data: Bucket 中包含的数据。
strlen($bucket->data): Bucket 中数据的长度。

五、实现一个简单的自定义流过滤器：ROT13 加密

下面我们来实现一个简单的自定义流过滤器，用于对数据进行 ROT13 加密。

<?php

class ROT13Filter extends php_user_filter
{
    public function filter($in, $out, &$consumed, $closing)
    {
        $consumed = 0;

        while ($bucket = stream_bucket_make_writeable($in)) {
            $bucket->data = str_rot13($bucket->data);
            $consumed += strlen($bucket->data);
            stream_bucket_append($out, $bucket);
        }

        return PSFS_PASS_ON;
    }
}

// 注册过滤器
stream_filter_register("rot13", "ROT13Filter");

// 使用过滤器
$file = fopen("test.txt", "r+");
stream_filter_append($file, "rot13");

// 读取文件内容并输出
while (!feof($file)) {
    echo fread($file, 8192);
}

fclose($file);

?>

代码解释：

ROT13Filter 类继承自 php_user_filter。
filter() 方法遍历输入 Bucket Brigade 中的每个 Bucket。
对于每个 Bucket，使用 str_rot13() 函数对数据进行 ROT13 加密。
将加密后的数据存储到新的 Bucket 中，并添加到输出 Bucket Brigade。
stream_filter_register("rot13", "ROT13Filter") 将过滤器类注册到 PHP 中，名称为 "rot13"。
stream_filter_append($file, "rot13") 将 "rot13" 过滤器附加到文件流 $file。
读取文件内容时，数据会经过 ROT13 过滤器加密后输出。
$consumed变量用于跟踪已处理的数据量。

六、实现一个更复杂的自定义流过滤器：解包固定长度记录

现在让我们考虑一个更复杂的场景：我们需要从一个二进制文件中读取固定长度的记录，并将每个记录解包成一个关联数组。假设每个记录包含以下字段：

字段	类型	长度 (字节)
`id`	整数	4
`name`	字符串	20
`amount`	浮点数	8

我们可以创建一个自定义流过滤器来处理这种数据格式。

<?php

class FixedLengthRecordFilter extends php_user_filter
{
    private $recordLength = 32; // 4 + 20 + 8
    private $fieldDefinitions = [
        ['name' => 'id', 'type' => 'int', 'length' => 4],
        ['name' => 'name', 'type' => 'string', 'length' => 20],
        ['name' => 'amount', 'type' => 'float', 'length' => 8],
    ];
    private $buffer = '';
    private $records = [];

    public function filter($in, $out, &$consumed, $closing)
    {
        $consumed = 0;

        while ($bucket = stream_bucket_make_writeable($in)) {
            $this->buffer .= $bucket->data;
            $bucketLength = strlen($bucket->data);
            $consumed += $bucketLength;

            // Process complete records from the buffer
            while (strlen($this->buffer) >= $this->recordLength) {
                $recordData = substr($this->buffer, 0, $this->recordLength);
                $this->buffer = substr($this->buffer, $this->recordLength);

                $record = $this->unpackRecord($recordData);
                $this->records[] = $record;

                // Create a new bucket for the unpacked record
                $newBucket = stream_bucket_new($this->stream, json_encode($record) . "n"); //JSON format
                stream_bucket_append($out, $newBucket);
            }
            stream_bucket_free($bucket); // Free the bucket after processing.
        }
        return PSFS_PASS_ON;
    }

    private function unpackRecord($recordData)
    {
        $record = [];
        $offset = 0;
        foreach ($this->fieldDefinitions as $field) {
            $fieldData = substr($recordData, $offset, $field['length']);
            $offset += $field['length'];

            switch ($field['type']) {
                case 'int':
                    $record[$field['name']] = unpack("i", $fieldData)[1];
                    break;
                case 'float':
                    $record[$field['name']] = unpack("d", $fieldData)[1];
                    break;
                case 'string':
                    $record[$field['name']] = trim($fieldData);
                    break;
            }
        }
        return $record;
    }
}

// 注册过滤器
stream_filter_register("fixed_length_record", "FixedLengthRecordFilter");

// 创建一个包含固定长度记录的二进制文件 (为了演示，手动创建并写入数据)
$binaryData = '';
$records = [
    ['id' => 1, 'name' => 'Alice', 'amount' => 100.50],
    ['id' => 2, 'name' => 'Bob', 'amount' => 200.75],
    ['id' => 3, 'name' => 'Charlie', 'amount' => 300.25],
];

foreach ($records as $record) {
    $binaryData .= pack("i", $record['id']);
    $binaryData .= str_pad($record['name'], 20, " ");
    $binaryData .= pack("d", $record['amount']);
}

file_put_contents("records.bin", $binaryData); // create binary file

// 使用过滤器读取二进制文件
$file = fopen("records.bin", "r");
stream_filter_append($file, "fixed_length_record");

// 读取文件内容并输出 (现在输出的是 JSON 格式的记录)
while (!feof($file)) {
    echo fread($file, 8192);
}

fclose($file);
?>

代码解释：

FixedLengthRecordFilter 类继承自 php_user_filter。
$recordLength 属性定义了每个记录的长度。
$fieldDefinitions 属性定义了每个字段的名称、类型和长度。
$buffer 属性用于存储从流中读取的数据，直到凑够一个完整的记录。
filter() 方法从输入 Bucket Brigade 中读取数据，并将其追加到 $buffer。
如果 $buffer 的长度大于等于 $recordLength，则从 $buffer 中提取一个完整的记录。
unpackRecord() 方法根据 $fieldDefinitions 解包记录中的每个字段。
将解包后的记录存储到 $records 数组中。
创建新的bucket，并将记录转成json字符串格式，追加到输出bucket brigade。
stream_bucket_free($bucket) 释放已经处理的bucket，避免内存泄漏。
代码中创建了一个包含固定长度记录的二进制文件，并使用 pack() 函数将数据打包成二进制格式。
最后，使用过滤器读取二进制文件，并将解包后的记录以 JSON 格式输出。

七、上下文管理

在一些情况下，我们需要在过滤器中维护一些状态信息，或者需要在多个过滤器实例之间共享数据。这时，可以使用流上下文（stream context）来实现上下文管理。

流上下文是一个资源类型，它包含一组参数和选项，可以传递给流操作函数，例如 fopen()、stream_socket_client() 等。我们可以使用 stream_context_create() 函数创建一个流上下文，并使用 stream_context_set_option() 函数设置上下文选项。

在自定义流过滤器中，可以通过 $this->stream 属性访问与过滤器关联的流资源。然后，可以使用 stream_get_meta_data() 函数获取流的元数据，其中包括流上下文。

八、一个使用上下文的例子
假设我们要实现一个过滤器，该过滤器可以根据上下文中的密钥对数据进行加密或解密。

<?php
class EncryptionFilter extends php_user_filter {
    private $key;
    private $encrypt; // true for encryption, false for decryption

    public function onCreate(): bool {
        $options = stream_context_get_options($this->stream);
        if (isset($options['encryption']['key'])) {
            $this->key = $options['encryption']['key'];
        } else {
            trigger_error('Encryption key is not set in the stream context.', E_USER_WARNING);
            return false;
        }

        if (isset($options['encryption']['encrypt'])) {
            $this->encrypt = (bool) $options['encryption']['encrypt'];
        } else {
            $this->encrypt = true; // Default to encryption
        }

        return true;
    }

    public function filter($in, $out, &$consumed, $closing): int {
        $consumed = 0;
        while ($bucket = stream_bucket_make_writeable($in)) {
            $data = $bucket->data;
            if ($this->encrypt) {
                $encrypted = $this->encryptData($data, $this->key);
                $bucket->data = $encrypted;
            } else {
                $decrypted = $this->decryptData($data, $this->key);
                $bucket->data = $decrypted;
            }
            $consumed += strlen($data);
            stream_bucket_append($out, $bucket);
        }
        return PSFS_PASS_ON;
    }

    private function encryptData(string $data, string $key): string {
        $result = '';
        $keyLength = strlen($key);
        for ($i = 0; $i < strlen($data); $i++) {
            $keyChar = $key[$i % $keyLength];
            $result .= chr(ord($data[$i]) + ord($keyChar)); // Simple encryption
        }
        return $result;
    }

    private function decryptData(string $data, string $key): string {
        $result = '';
        $keyLength = strlen($key);
        for ($i = 0; $i < strlen($data); $i++) {
            $keyChar = $key[$i % $keyLength];
            $result .= chr(ord($data[$i]) - ord($keyChar)); // Simple decryption
        }
        return $result;
    }
}

stream_filter_register('encryption', 'EncryptionFilter');

// Example Usage
$key = 'mysecretkey';
$data = 'This is some sensitive data.';

// Encryption
$context = stream_context_create([
    'encryption' => [
        'key' => $key,
        'encrypt' => true, // Explicitly set to encrypt
    ],
]);

$stream = fopen('php://memory', 'r+', false, $context);
stream_filter_append($stream, 'encryption');
fwrite($stream, $data);
rewind($stream);
$encryptedData = stream_get_contents($stream);
fclose($stream);

echo "Encrypted Data: " . base64_encode($encryptedData) . "n";  // Output the base64 encoded encrypted data

// Decryption
$context = stream_context_create([
    'encryption' => [
        'key' => $key,
        'encrypt' => false, // Set to decrypt
    ],
]);

$stream = fopen('php://memory', 'r+', false, $context);
stream_filter_append($stream, 'encryption');
fwrite($stream, base64_decode($encryptedData)); // Write the base64 decoded encrypted data
rewind($stream);
$decryptedData = stream_get_contents($stream);
fclose($stream);

echo "Decrypted Data: " . $decryptedData . "n"; // Should output the original data
?>

代码解释:

EncryptionFilter 类从流上下文中获取加密密钥和加密/解密标志。
在 onCreate() 方法中，它使用 stream_context_get_options() 来检索上下文选项。如果密钥未设置，则会触发错误。
filter() 方法使用密钥和加密/解密标志来加密或解密数据。
示例用法演示了如何创建带有上下文选项的流，附加加密过滤器，写入数据，然后读取加密的数据。
解密过程类似，但上下文选项设置为 encrypt => false。

九、用户空间协议实现

自定义流过滤器可以用于实现用户空间协议。用户空间协议是指在应用程序层实现的协议，而不是在操作系统内核中实现的协议。

例如，我们可以使用自定义流过滤器来实现一个简单的 HTTP 客户端，用于从服务器接收数据，并对数据进行解析。

十、总结
自定义流过滤器是PHP中一个强大的工具，允许在数据流的处理过程中进行灵活的修改和转换。它们可以用于各种任务，比如数据加密/解密、数据格式转换、以及实现用户空间协议。通过理解 php_user_filter 类、Bucket Brigade 的使用、以及流上下文的管理，可以充分利用流过滤器的功能。掌握这些技术点能够帮助开发者更有效地处理数据流，构建更加健壮和灵活的应用程序。

PHP 自定义流过滤器：上下文管理与用户空间协议实现

发表回复 取消回复

发表回复取消回复