blob: de0866f691f782756de666d5054681374bee1a97 [file]
<?php
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
declare(strict_types=1);
use Iggy\Client as IggyClient;
use Iggy\ReceiveMessage;
use PHPUnit\Framework\Assert;
function assert_true(bool $condition, string $message = 'expected condition to be true'): void
{
Assert::assertTrue($condition, $message);
}
function assert_same(mixed $expected, mixed $actual, string $message = ''): void
{
Assert::assertSame($expected, $actual, $message);
}
function assert_not_null(mixed $value, string $message = 'expected value not to be null'): void
{
Assert::assertNotNull($value, $message);
}
function assert_instance_of(string $expected, mixed $actual, string $message = ''): void
{
Assert::assertInstanceOf($expected, $actual, $message);
}
function assert_null(mixed $value, string $message = 'expected value to be null'): void
{
Assert::assertNull($value, $message);
}
function assert_count(int $expected, array $actual, string $message = ''): void
{
Assert::assertCount($expected, $actual, $message);
}
function assert_throws(callable $callback, ?string $messageContains = null): Throwable
{
try {
$callback();
} catch (Throwable $throwable) {
if ($messageContains !== null && !str_contains($throwable->getMessage(), $messageContains)) {
Assert::fail(
'expected exception message to contain ' . var_export($messageContains, true)
. ', got ' . var_export($throwable->getMessage(), true)
);
}
Assert::assertInstanceOf(Throwable::class, $throwable);
return $throwable;
}
Assert::fail('expected callable to throw');
}
function unique_name(string $prefix): string
{
return $prefix . '-' . bin2hex(random_bytes(4));
}
function env_or_default(string $name, string $default): string
{
$value = getenv($name);
return $value === false || $value === '' ? $default : $value;
}
function server_host(): string
{
return env_or_default('IGGY_HOST', '127.0.0.1');
}
function server_port(): int
{
return (int) env_or_default('IGGY_PORT', '8090');
}
function wait_for_server(string $host, int $port, int $timeoutSeconds = 30): void
{
$deadline = microtime(true) + $timeoutSeconds;
$lastError = null;
while (microtime(true) < $deadline) {
$socket = @fsockopen($host, $port, $errno, $errstr, 1.0);
if (is_resource($socket)) {
fclose($socket);
return;
}
$lastError = trim($errstr !== '' ? $errstr : (string) $errno);
usleep(250_000);
}
Assert::fail("Iggy server was not reachable at {$host}:{$port}" . ($lastError !== null ? " ({$lastError})" : ''));
}
function new_client(): IggyClient
{
$client = new IggyClient(server_host() . ':' . server_port());
$client->connect();
$client->loginUser(env_or_default('IGGY_USERNAME', 'iggy'), env_or_default('IGGY_PASSWORD', 'iggy'));
return $client;
}
function new_connection_string_client(): IggyClient
{
$host = server_host();
$port = server_port();
$username = rawurlencode(env_or_default('IGGY_USERNAME', 'iggy'));
$password = rawurlencode(env_or_default('IGGY_PASSWORD', 'iggy'));
$client = IggyClient::fromConnectionString("iggy+tcp://{$username}:{$password}@{$host}:{$port}");
$client->connect();
return $client;
}
function create_stream_and_topic(object $client, string $stream, string $topic, int $partitions = 1): void
{
$client->createStream($stream);
$client->createTopic($stream, $topic, $partitions, null, null, null, null);
}
function cleanup_topic(object $client, string|int $stream, string|int $topic): void
{
try {
$client->deleteTopic($stream, $topic);
} catch (Throwable) {
}
}
function cleanup_stream(object $client, string|int $stream): void
{
try {
$client->deleteStream($stream);
} catch (Throwable) {
}
}
function cleanup_stream_with_topics(object $client, string|int $stream, array $topics): void
{
foreach ($topics as $topic) {
cleanup_topic($client, $stream, $topic);
}
cleanup_stream($client, $stream);
}
function collect_payloads(array $messages): array
{
return array_map(static fn (ReceiveMessage $message): string => $message->payload(), $messages);
}
function collect_offsets(array $messages): array
{
return array_map(static fn (ReceiveMessage $message): int => $message->offset(), $messages);
}
function micros(int $seconds): int
{
return $seconds * 1_000_000;
}
if (!extension_loaded('iggy-php')) {
Assert::fail('The iggy-php extension is not loaded.');
}
wait_for_server(server_host(), server_port());