Update to latest websocket changes

composer.json

@@ -33,24 +33,22 @@
  "amphp/amp": "^3",
  "amphp/byte-stream": "^2",
  "amphp/http": "^2",
- "amphp/http-client": "^5",
+ "amphp/http-client": "^5-dev",
  "amphp/socket": "^2",
- "amphp/websocket": "^2",
+ "amphp/websocket": "^2-dev",
  "league/uri": "^6",
  "psr/http-message": "^1",
  "revolt/event-loop": "^1"
  },
  "require-dev": {
  "amphp/http-server": "^3",
- "amphp/websocket-server": "^3",
+ "amphp/websocket-server": "^3-dev",
  "amphp/phpunit-util": "^3",
  "amphp/php-cs-fixer-config": "^2",
  "phpunit/phpunit": "^9",
  "psr/log": "^1",
  "psalm/phar": "^5.4"
  },
- "minimum-stability": "dev",
- "prefer-stable": true,
  "autoload": {
  "psr-4": {
  "Amp\\Websocket\\Client\\": "src"

docs/index.md

@@ -71,7 +71,7 @@ use Amp\Websocket\Client;
 Amp\Loop::run(function () {
  /** @var Client\WebsocketConnection $connection */
  $connection = yield Client\connect('ws://demos.kaazing.com/echo');
- yield $connection->send('Hello!');
+ yield $connection->sendText('Hello!');
 
  $i = 0;
 
@@ -89,9 +89,9 @@ Amp\Loop::run(function () {
  yield new Delayed(1000);
 
  if ($i < 3) {
- yield $connection->send('Ping: ' . ++$i);
+ yield $connection->sendText('Ping: ' . ++$i);
  } else {
- yield $connection->send('Goodbye!');
+ yield $connection->sendText('Goodbye!');
  }
  }
 });

examples/amp.php

@@ -13,7 +13,7 @@
 
 $connection = connect($handshake);
 
-$connection->send('Hello!');
+$connection->sendText('Hello!');
 
 $i = 0;
 
@@ -30,8 +30,8 @@
  delay(1);
 
  if ($i < 3) {
- $connection->send('Ping: ' . ++$i);
+ $connection->sendText('Ping: ' . ++$i);
  } else {
- $connection->send('Goodbye!');
+ $connection->sendText('Goodbye!');
  }
 }

src/Rfc6455Connection.php

@@ -9,12 +9,14 @@
 use Amp\Http\Client\Response;
 use Amp\Socket\SocketAddress;
 use Amp\Socket\TlsInfo;
-use Amp\Websocket\CloseCode;
 use Amp\Websocket\Rfc6455Client;
-use Amp\Websocket\WebsocketClientMetadata;
+use Amp\Websocket\WebsocketCloseCode;
+use Amp\Websocket\WebsocketCount;
 use Amp\Websocket\WebsocketMessage;
+use Amp\Websocket\WebsocketTimestamp;
+use Traversable;
 
-final class Rfc6455Connection implements WebsocketConnection
+final class Rfc6455Connection implements WebsocketConnection, \IteratorAggregate
 {
  use ForbidCloning;
  use ForbidSerialization;
@@ -63,11 +65,6 @@ public function isClosedByPeer(): bool
  return $this->client->isClosedByPeer();
  }
 
- public function getUnansweredPingCount(): int
- {
- return $this->client->getUnansweredPingCount();
- }
-
  public function getCloseCode(): int
  {
  return $this->client->getCloseCode();
@@ -78,19 +75,19 @@ public function getCloseReason(): string
  return $this->client->getCloseReason();
  }
 
- public function send(string $data): void
+ public function sendText(string $data): void
  {
- $this->client->send($data);
+ $this->client->sendText($data);
  }
 
  public function sendBinary(string $data): void
  {
  $this->client->sendBinary($data);
  }
 
- public function stream(ReadableStream $stream): void
+ public function streamText(ReadableStream $stream): void
  {
- $this->client->stream($stream);
+ $this->client->streamText($stream);
  }
 
  public function streamBinary(ReadableStream $stream): void
@@ -103,17 +100,22 @@ public function ping(): void
  $this->client->ping();
  }
 
- public function getInfo(): WebsocketClientMetadata
+ public function getCount(WebsocketCount $type): int
+ {
+ return $this->client->getCount($type);
+ }
+
+ public function getTimestamp(WebsocketTimestamp $type): int
  {
- return $this->client->getInfo();
+ return $this->client->getTimestamp($type);
  }
 
  public function isClosed(): bool
  {
  return $this->client->isClosed();
  }
 
- public function close(int $code = CloseCode::NORMAL_CLOSE, string $reason = ''): void
+ public function close(int $code = WebsocketCloseCode::NORMAL_CLOSE, string $reason = ''): void
  {
  $this->client->close($code, $reason);
  }
@@ -122,4 +124,14 @@ public function onClose(\Closure $onClose): void
  {
  $this->client->onClose($onClose);
  }
+
+ public function isCompressionEnabled(): bool
+ {
+ return $this->client->isCompressionEnabled();
+ }
+
+ public function getIterator(): Traversable
+ {
+ yield from $this->client;
+ }
 }

src/Rfc6455ConnectionFactory.php

@@ -6,21 +6,21 @@
 use Amp\ForbidSerialization;
 use Amp\Http\Client\Response;
 use Amp\Socket\Socket;
-use Amp\Websocket\Compression\CompressionContext;
-use Amp\Websocket\HeartbeatQueue;
+use Amp\Websocket\Compression\WebsocketCompressionContext;
 use Amp\Websocket\Parser\Rfc6455ParserFactory;
 use Amp\Websocket\Parser\WebsocketParserFactory;
-use Amp\Websocket\RateLimiter;
 use Amp\Websocket\Rfc6455Client;
+use Amp\Websocket\WebsocketHeartbeatQueue;
+use Amp\Websocket\WebsocketRateLimit;
 
 final class Rfc6455ConnectionFactory implements WebsocketConnectionFactory
 {
  use ForbidCloning;
  use ForbidSerialization;
 
  public function __construct(
- private readonly ?HeartbeatQueue $heartbeatQueue = null,
- private readonly ?RateLimiter $rateLimiter = null,
+ private readonly ?WebsocketHeartbeatQueue $heartbeatQueue = null,
+ private readonly ?WebsocketRateLimit $rateLimit = null,
  private readonly WebsocketParserFactory $parserFactory = new Rfc6455ParserFactory(
  messageSizeLimit: Rfc6455Connection::DEFAULT_MESSAGE_SIZE_LIMIT,
  frameSizeLimit: Rfc6455Connection::DEFAULT_FRAME_SIZE_LIMIT,
@@ -33,15 +33,15 @@ public function __construct(
  public function createConnection(
  Response $handshakeResponse,
  Socket $socket,
- ?CompressionContext $compressionContext = null,
+ ?WebsocketCompressionContext $compressionContext = null,
  ): WebsocketConnection {
  $client = new Rfc6455Client(
  socket: $socket,
  masked: true,
  parserFactory: $this->parserFactory,
  compressionContext: $compressionContext,
  heartbeatQueue: $this->heartbeatQueue,
- rateLimiter: $this->rateLimiter,
+ rateLimit: $this->rateLimit,
  frameSplitThreshold: $this->frameSplitThreshold,
  closePeriod: $this->closePeriod,
  );

src/Rfc6455Connector.php

@@ -16,8 +16,8 @@
 use Amp\Socket\ConnectContext;
 use Amp\Socket\Socket;
 use Amp\Websocket;
-use Amp\Websocket\Compression\CompressionContextFactory;
 use Amp\Websocket\Compression\Rfc7692CompressionFactory;
+use Amp\Websocket\Compression\WebsocketCompressionContextFactory;
 
 final class Rfc6455Connector implements WebsocketConnector
 {
@@ -27,12 +27,12 @@ final class Rfc6455Connector implements WebsocketConnector
  private readonly HttpClient $httpClient;
 
  /**
- * @param CompressionContextFactory|null $compressionContextFactory Use null to disable compression.
+ * @param WebsocketCompressionContextFactory|null $compressionContextFactory Use null to disable compression.
  */
  public function __construct(
  private readonly WebsocketConnectionFactory $connectionFactory = new Rfc6455ConnectionFactory(),
  HttpClient $httpClient = null,
- private readonly ?CompressionContextFactory $compressionContextFactory = new Rfc7692CompressionFactory(),
+ private readonly ?WebsocketCompressionContextFactory $compressionContextFactory = new Rfc7692CompressionFactory(),
  ) {
  $this->httpClient = $httpClient
  ?? (new HttpClientBuilder)->usingPool(

src/WebsocketConnectionFactory.php

@@ -4,18 +4,18 @@
 
 use Amp\Http\Client\Response;
 use Amp\Socket\Socket;
-use Amp\Websocket\Compression\CompressionContext;
+use Amp\Websocket\Compression\WebsocketCompressionContext;
 
 interface WebsocketConnectionFactory
 {
  /**
  * @param Response $handshakeResponse Response that initiated the websocket connection.
  * @param Socket $socket Underlying socket to be used for network communication.
- * @param CompressionContext|null $compressionContext CompressionContext generated from the response headers.
+ * @param WebsocketCompressionContext|null $compressionContext CompressionContext generated from the response headers.
  */
  public function createConnection(
  Response $handshakeResponse,
  Socket $socket,
- ?CompressionContext $compressionContext = null,
+ ?WebsocketCompressionContext $compressionContext = null,
  ): WebsocketConnection;
 }

test-autobahn/runner.php

@@ -4,8 +4,8 @@
 use Amp\Websocket\Client\Rfc6455ConnectionFactory;
 use Amp\Websocket\Client\Rfc6455Connector;
 use Amp\Websocket\Client\WebsocketHandshake;
-use Amp\Websocket\ClosedException;
 use Amp\Websocket\Parser\Rfc6455ParserFactory;
+use Amp\Websocket\WebsocketClosedException;
 
 require __DIR__ . '/../vendor/autoload.php';
 
@@ -42,10 +42,10 @@
  if ($message->isBinary()) {
  $connection->sendBinary($content);
  } else {
- $connection->send($content);
+ $connection->sendText($content);
  }
  }
- } catch (ClosedException $e) {
+ } catch (WebsocketClosedException $e) {
  // ignore
  } catch (AssertionError $e) {
  print 'Assertion error: ' . $e->getMessage() . PHP_EOL;

test/WebsocketConnectionTest.php

@@ -13,12 +13,12 @@
 use Amp\Socket\SocketException;
 use Amp\TimeoutCancellation;
 use Amp\Websocket\Client;
-use Amp\Websocket\ClosedException;
 use Amp\Websocket\Parser\Rfc6455ParserFactory;
 use Amp\Websocket\Server\EmptyWebsocketHandshakeHandler;
 use Amp\Websocket\Server\Websocket;
 use Amp\Websocket\Server\WebsocketClientHandler;
 use Amp\Websocket\WebsocketClient;
+use Amp\Websocket\WebsocketClosedException;
 use Psr\Log\NullLogger;
 use function Amp\async;
 use function Amp\delay;
@@ -82,14 +82,14 @@ public function testSimpleTextEcho(): void
  public function handleClient(WebsocketClient $client, Request $request, Response $response): void
  {
  while ($message = $client->receive()) {
- $client->send($message->buffer());
+ $client->sendText($message->buffer());
  }
  }
  });
 
  try {
  $client = connect('ws://' . $address->toString());
- $client->send('Hey!');
+ $client->sendText('Hey!');
 
  $message = $client->receive();
 
@@ -110,8 +110,8 @@ public function testUnconsumedMessage(): void
  [$server, $address] = $this->createServer(new class implements WebsocketClientHandler {
  public function handleClient(WebsocketClient $client, Request $request, Response $response): void
  {
- $client->send(\str_repeat('.', 1024 * 1024));
- $client->send('Message');
+ $client->sendText(\str_repeat('.', 1024 * 1024));
+ $client->sendText('Message');
  $client->close();
  }
  });
@@ -188,7 +188,7 @@ public function handleClient(WebsocketClient $client, Request $request, Response
  $message->buffer();
 
  self::fail('Buffering the message should have thrown a ClosedException due to exceeding the message size limit');
- } catch (ClosedException $exception) {
+ } catch (WebsocketClosedException $exception) {
  $this->assertSame('Received payload exceeds maximum allowable size', $exception->getReason());
  } finally {
  $server->stop();