| <?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()); |