Implement realtime API

composer.json

@@ -11,6 +11,7 @@
     "xp-framework/reflection": "^3.0 | ^2.0",
     "xp-forge/marshalling": "^2.0 | ^1.0",
     "xp-forge/rest-client": "^5.6",
+    "xp-forge/websockets": "^4.0",
     "php" : ">=7.4.0"
   },
   "require-dev" : {

src/main/php/com/openai/realtime/RealtimeApi.class.php

@@ -0,0 +1,103 @@
+<?php namespace com\openai\realtime;
+
+use lang\IllegalStateException;
+use text\json\Json;
+use util\data\Marshalling;
+use util\log\Traceable;
+use util\URI;
+use websocket\WebSocket;
+
+/**
+ * OpenAI Realtime API enables you to build low-latency, multi-modal conversational
+ * experiences. It currently supports text and audio as both input and output, as
+ * well as function calling.
+ *
+ * @test  com.openai.unittest.RealtimeApiTest
+ * @see   https://platform.openai.com/docs/guides/realtime
+ */
+class RealtimeApi implements Traceable {
+  private $ws, $marshalling;
+  private $cat= null;
+
+  /** @param string|util.URI|websocket.WebSocket $endpoint */
+  public function __construct($endpoint) {
+    $this->ws= $endpoint instanceof WebSocket ? $endpoint : new WebSocket((string)$endpoint);
+    $this->marshalling= new Marshalling();
+  }
+
+  /** @param ?util.log.LogCategory $cat */
+  public function setTrace($cat) {
+    $this->cat= $cat;
+  }
+
+  /**
+   * Opens the underlying websocket, optionally passing headers
+   *
+   * Verifies a `session.created` event is received. This is sent by the server
+   * as soon as the connection is successfully established. Provides a connection-
+   * specific ID that may be useful for debugging or logging.
+   *
+   * @return var
+   * @throws lang.IllegalStateException
+   */
+  public function connect(array $headers= []) {
+    $this->cat && $this->cat->info($this->ws->socket(), $this->ws->path(), $headers);
+    $this->ws->connect($headers);
+
+    $event= $this->receive();
+    if ('session.created' === ($event['type'] ?? null)) return $event;
+
+    $error= 'Unexpected handshake event "'.($event['type'] ?? '(null)').'"';
+    $this->ws->close(4007, $error);
+    throw new IllegalStateException($error);
+  }
+
+  /** Returns whether the underlying websocket is connected */
+  public function connected(): bool {
+    return $this->ws->connected();
+  }
+
+  /** Closes the underlying websocket */
+  public function close(): void {
+    $this->ws->close();
+  }
+
+  /**
+   * Sends a given payload. Doesn't wait for a response
+   *
+   * @param  var $payload
+   * @return void
+   */
+  public function send($payload): void {
+    $json= Json::of($this->marshalling->marshal($payload));
+    $this->cat && $this->cat->debug('>>>', $json);
+    $this->ws->send($json);
+  }
+
+  /**
+   * Receives an answer. Returns NULL if EOF is reached.
+   *
+   * @return var
+   */
+  public function receive() {
+    $json= $this->ws->receive();
+    $this->cat && $this->cat->debug('<<<', $json);
+    return null === $json ? null : $this->marshalling->unmarshal(Json::read($json));
+  }
+
+  /**
+   * Sends a given payload and returns the response to it.
+   *
+   * @param  var $payload
+   * @return var
+   */
+  public function transmit($payload) {
+    $this->send($payload);
+    return $this->receive();
+  }
+
+  /** Ensures socket is closed */
+  public function __destruct() {
+    $this->ws && $this->ws->close();
+  }
+}

src/test/php/com/openai/unittest/RealtimeApiTest.class.php

@@ -0,0 +1,94 @@
+<?php namespace com\openai\unittest;
+
+use com\openai\realtime\RealtimeApi;
+use lang\IllegalStateException;
+use test\{Assert, Expect, Test, Values};
+
+class RealtimeApiTest {
+  const SESSION_CREATED= '{"type": "session.created"}';
+
+  /** Returns authentications */
+  private function authentications(): iterable {
+    yield ['azure', ['api-key' => 'test']];
+    yield ['openai', ['Authorization' => 'Bearer test', 'OpenAI-Beta' => 'realtime=v1']];
+  }
+
+  #[Test]
+  public function can_create() {
+    new RealtimeApi('wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01');
+  }
+
+  #[Test]
+  public function initially_not_connected() {
+    $c= new RealtimeApi(new TestingSocket());
+
+    Assert::false($c->connected());
+  }
+
+  #[Test]
+  public function connect() {
+    $c= new RealtimeApi(new TestingSocket([self::SESSION_CREATED]));
+    $c->connect();
+
+    Assert::true($c->connected());
+  }
+
+  #[Test, Values(from: 'authentications')]
+  public function passing_headers($kind, $headers) {
+    $s= new TestingSocket([self::SESSION_CREATED]);
+
+    $c= new RealtimeApi($s);
+    $c->connect($headers);
+
+    Assert::equals($headers, $s->connected);
+  }
+
+  #[Test]
+  public function close() {
+    $c= new RealtimeApi(new TestingSocket([self::SESSION_CREATED]));
+    $c->connect();
+    $c->close();
+
+    Assert::false($c->connected());
+  }
+
+  #[Test]
+  public function initial_handshake() {
+    $c= new RealtimeApi(new TestingSocket([self::SESSION_CREATED]));
+    $session= $c->connect();
+
+    Assert::equals(['type' => 'session.created'], $session);
+  }
+
+  #[Test, Expect(class: IllegalStateException::class, message: 'Unexpected handshake event "error"')]
+  public function unexpected_handshake() {
+    $c= new RealtimeApi(new TestingSocket(['{"type":"error"}']));
+    $c->connect();
+  }
+
+  #[Test]
+  public function update_session() {
+    $c= new RealtimeApi(new TestingSocket([
+      self::SESSION_CREATED,
+      '{"type": "session.update", "session": {"instructions": "You are TestGPT"}}',
+      '{"type": "session.updated"}',
+    ]));
+    $c->connect();
+    $c->send(['type' => 'session.update', 'session' => ['instructions' => 'You are TestGPT']]);
+
+    Assert::equals(['type' => 'session.updated'], $c->receive());
+  }
+
+  #[Test]
+  public function transmit() {
+    $c= new RealtimeApi(new TestingSocket([
+      self::SESSION_CREATED,
+      '{"type": "conversation.item.create", "item": {"type": "message"}}',
+      '{"type": "conversation.item.created"}',
+    ]));
+    $c->connect();
+    $response= $c->transmit(['type' => 'conversation.item.create', 'item' => ['type' => 'message']]);
+
+    Assert::equals(['type' => 'conversation.item.created'], $response);
+  }
+}

src/test/php/com/openai/unittest/TestingSocket.class.php

@@ -0,0 +1,36 @@
+<?php namespace com\openai\unittest;
+
+use lang\IllegalStateException;
+use websocket\WebSocket;
+
+class TestingSocket extends WebSocket {
+  private $messages;
+  public $connected= null;
+
+  public function __construct($messages= []) {
+    $this->messages= $messages;
+  }
+
+  public function connected() {
+    return isset($this->connected);
+  }
+
+  public function connect($headers= []) {
+    $this->connected= $headers;
+  }
+
+  public function send($payload) {
+    $message= array_shift($this->messages);
+    if (json_decode($message, true) !== json_decode($payload, true)) {
+      throw new IllegalStateException('Unexpected '.$payload.', expecting '.$message);
+    }
+  }
+
+  public function receive($timeout= null) {
+    return array_shift($this->messages);
+  }
+
+  public function close($code= 1000, $reason = '') {
+    $this->connected= null;
+  }
+}