thekid/Chat.php

## chat.html
<!DOCTYPE html>
<html>
  <head>
    <meta charset="utf-8"/>
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta name="msapplication-config" content="none"/>
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <title>WebSocket test</title>
    <link rel="stylesheet" type="text/css" href="//cdn.jsdelivr.net/npm/fomantic-ui@2.7.7/dist/semantic.min.css">
  </head>
  <body class="pushable">
    <div class="pusher">
      <div id="status" class="ui inverted clearing segment">Disconnected</div>
      <div class="ui vertical segment">
        <div class="ui form">
          <div class="field">
            <label>Channel</label>
            <select id="channel" class="ui search dropdown" onchange="join(this)">
              <option value="announcements">Announcements</option>
              <option value="karlsruhe">#Karlsruhe</option>
              <option value="agile">#Agile</option>
            </select>
          </div>
          <div class="field">
            <textarea id="entry" cols="76" rows="3" onkeydown="return send(this);" disabled></textarea>
          </div>
        </div>
      </div>
      <div class="ui segment">
        <div class="event" id="event-template" style="display: none">
          <div class="content">
            <div class="summary">
              <span class="type">{{type}}</span>
              <div class="date">{{date}}</div>
            </div>
            <div class="extra text" style="white-space: pre">{{content}}</div>
          </div>
        </div>

        <div class="ui large feed" id="response">
          <!-- Nodes will be inserted here -->
        </div>
      </div>
    </div>
    <script src="//cdn.jsdelivr.net/npm/jquery@3.3.1/dist/jquery.min.js"></script>
    <script src="//cdn.jsdelivr.net/npm/fomantic-ui@2.7.7/dist/semantic.min.js"></script>
    <script type="text/javascript">
      var conn = null;
      var channel = null;

      function add(type, text) {
        var $response = document.getElementById('response');
        var $node = document.getElementById('event-template').cloneNode(true);

        $node.querySelector('.content .type').innerText = type;
        $node.querySelector('.content .date').innerText = new Date().toString();
        $node.querySelector('.content .text').innerText = text;
        $node.style.display = 'block';
        $response.insertBefore($node, $response.childNodes[0] || null);
      }

      function disconnect() {
        var $status = document.getElementById('status');

        if (conn) {
          clearInterval(conn.interval);
          conn.socket.close(3000);
          conn = null;
        }
      }

      function connect() {
        var $status = document.getElementById('status');
        var $entry = document.getElementById('entry');

        disconnect();
        $status.classList.remove('red');
        $status.classList.remove('green');
        $status.innerText = 'Connecting...';

        var ws = new WebSocket("ws://localhost:8081/chat");
        ws.onerror = function(error) {
          console.log(error);
          $status.classList.add('red');
          $status.innerHTML = '<button class="ui tiny right floated button" onclick="connect()">Try again</button>';
        };
        ws.onopen = function() {
          add('Connected', 'Start typing, hit Ctrl+Enter to send!');
          $entry.disabled = false;
          $entry.focus();
          $status.classList.add('green');
          $status.innerHTML = '<button class="ui tiny right floated button" onclick="disconnect()">Disconnect</button>';

          // Join selected channel
          join(document.getElementById('channel'));
        };
        ws.onmessage = function(event) {
          var message = JSON.parse(event.data);
          switch (message.kind) {
            case 'ACCEPTED': case 'PONG': break; // Noop
            case 'JOINED': add('Channel', 'You are now on ' + message.channel); channel = message.channel; break;
            case 'MESSAGE': add('Message', message.value); break;
            case 'ERROR': add('Error', message.value); break;
          }
        };
        ws.onclose = function(event) {
          if (event.code > 1000 && event.code < 3000) {
            add('Disconnected', 'Timeout reached, it seems (code #' + event.code + ')');
            $status.classList.remove('green');
            $status.classList.add('red');
          } else {
            add('Disconnected', 'Connection closed');
            $status.classList.remove('red');
            $status.classList.remove('green');
          }
          $status.innerHTML = '<button class="ui tiny right floated button" onclick="connect()">Reconnect</button>';
          $entry.disabled = true;
        };

        conn = {
          socket   : ws,
          interval : setInterval(function() { ws.send('{"kind":"PING"}'); }, 9 * 60 * 1000)
        };
      }

      function send(entry) {
        if ((window.event.keyCode == 10 || window.event.keyCode == 13) && window.event.ctrlKey) {
          try {
            conn.socket.send(JSON.stringify({kind: 'MESSAGE', channel: channel, message: entry.value}));
            entry.value = '';
          } catch (error) {
            console.log(error);
          }
          return false;
        }
      }

      function join(select) {
        conn.socket.send(JSON.stringify({kind: 'SUBSCRIBE', channel: select.value}));
      }

      // Establish connection
      connect();

      // Dropdown menu initialization
      $('#channel').dropdown();
    </script>
  </body>
</html>

## Chat.php
<?php

use io\redis\RedisProtocol;
use websocket\Listeners;

class Chat extends Listeners {
  private $subscriptions= [];

  /** @return [:var] */
  public function serve($events) {
    $dsn= $this->environment->arguments()[0] ?? 'redis://localhost';

    // Subscribe to Redis queues
    $sub= new RedisProtocol($dsn)->connect();
    $events->add($sub->socket(), fn() => {
      [$type, $channel, $message]= $sub->receive();
      foreach ($this->subscriptions[$channel] ?? [] as $index => $connection) {
        try {
          $connection->send(json_encode(['kind' => 'MESSAGE', 'value' => $message]));
        } catch ($e) {
          unset($this->subscriptions[$channel][$index]);  // Client disconnected
        }
      }
    });

    // Handle websocket messages
    $pub= new RedisProtocol($dsn)->connect();
    return [
      '/chat' => fn($connection, $message) => {
        $value= json_decode($message, true);
        $return= switch ($value['kind']) {
          case 'PING' => ['kind' => 'PONG'];

          case 'MESSAGE' => {
            $pub->command('PUBLISH', $value['channel'], $value['message']);
            return ['kind' => 'ACCEPTED'];
          }

          case 'SUBSCRIBE' => {
            foreach ($this->subscriptions as $channel => &$subscriptions) {
              unset($subscriptions[$connection->id()]);
              empty($subscriptions) && $sub->command('UNSUBSCRIBE', $channel);
            }
            $channel= $value['channel'];
            if (!isset($this->subscriptions[$channel])) {
              $sub->command('SUBSCRIBE', $channel);
              $this->subscriptions[$channel]= [];
            }
            $this->subscriptions[$channel][$connection->id()]= $connection;
            return ['kind' => 'JOINED', 'channel' => $channel];
          }

          default => ['kind' => 'ERROR', 'value' => 'Unknown '.$value['kind']];
        };

        $connection->send(json_encode($return));
        return $return['kind'];
      }
    ];
  }
}
	<!DOCTYPE html>
	<html>
	<head>
	<meta charset="utf-8"/>
	<meta name="viewport" content="width=device-width, initial-scale=1.0">
	<meta name="msapplication-config" content="none"/>
	<meta http-equiv="X-UA-Compatible" content="IE=edge">
	<title>WebSocket test</title>
	<link rel="stylesheet" type="text/css" href="//cdn.jsdelivr.net/npm/fomantic-ui@2.7.7/dist/semantic.min.css">
	</head>
	<body class="pushable">
	<div class="pusher">
	<div id="status" class="ui inverted clearing segment">Disconnected</div>
	<div class="ui vertical segment">
	<div class="ui form">
	<div class="field">
	<label>Channel</label>
	<select id="channel" class="ui search dropdown" onchange="join(this)">
	<option value="announcements">Announcements</option>
	<option value="karlsruhe">#Karlsruhe</option>
	<option value="agile">#Agile</option>
	</select>
	</div>
	<div class="field">
	<textarea id="entry" cols="76" rows="3" onkeydown="return send(this);" disabled></textarea>
	</div>
	</div>
	</div>
	<div class="ui segment">
	<div class="event" id="event-template" style="display: none">
	<div class="content">
	<div class="summary">
	<span class="type">{{type}}</span>
	<div class="date">{{date}}</div>
	</div>
	<div class="extra text" style="white-space: pre">{{content}}</div>
	</div>
	</div>

	<div class="ui large feed" id="response">
	<!-- Nodes will be inserted here -->
	</div>
	</div>
	</div>
	<script src="//cdn.jsdelivr.net/npm/jquery@3.3.1/dist/jquery.min.js"></script>
	<script src="//cdn.jsdelivr.net/npm/fomantic-ui@2.7.7/dist/semantic.min.js"></script>
	<script type="text/javascript">
	var conn = null;
	var channel = null;

	function add(type, text) {
	var $response = document.getElementById('response');
	var $node = document.getElementById('event-template').cloneNode(true);

	$node.querySelector('.content .type').innerText = type;
	$node.querySelector('.content .date').innerText = new Date().toString();
	$node.querySelector('.content .text').innerText = text;
	$node.style.display = 'block';
	$response.insertBefore($node, $response.childNodes[0] \|\| null);
	}

	function disconnect() {
	var $status = document.getElementById('status');

	if (conn) {
	clearInterval(conn.interval);
	conn.socket.close(3000);
	conn = null;
	}
	}

	function connect() {
	var $status = document.getElementById('status');
	var $entry = document.getElementById('entry');

	disconnect();
	$status.classList.remove('red');
	$status.classList.remove('green');
	$status.innerText = 'Connecting...';

	var ws = new WebSocket("ws://localhost:8081/chat");
	ws.onerror = function(error) {
	console.log(error);
	$status.classList.add('red');
	$status.innerHTML = '<button class="ui tiny right floated button" onclick="connect()">Try again</button>';
	};
	ws.onopen = function() {
	add('Connected', 'Start typing, hit Ctrl+Enter to send!');
	$entry.disabled = false;
	$entry.focus();
	$status.classList.add('green');
	$status.innerHTML = '<button class="ui tiny right floated button" onclick="disconnect()">Disconnect</button>';

	// Join selected channel
	join(document.getElementById('channel'));
	};
	ws.onmessage = function(event) {
	var message = JSON.parse(event.data);
	switch (message.kind) {
	case 'ACCEPTED': case 'PONG': break; // Noop
	case 'JOINED': add('Channel', 'You are now on ' + message.channel); channel = message.channel; break;
	case 'MESSAGE': add('Message', message.value); break;
	case 'ERROR': add('Error', message.value); break;
	}
	};
	ws.onclose = function(event) {
	if (event.code > 1000 && event.code < 3000) {
	add('Disconnected', 'Timeout reached, it seems (code #' + event.code + ')');
	$status.classList.remove('green');
	$status.classList.add('red');
	} else {
	add('Disconnected', 'Connection closed');
	$status.classList.remove('red');
	$status.classList.remove('green');
	}
	$status.innerHTML = '<button class="ui tiny right floated button" onclick="connect()">Reconnect</button>';
	$entry.disabled = true;
	};

	conn = {
	socket : ws,
	interval : setInterval(function() { ws.send('{"kind":"PING"}'); }, 9 * 60 * 1000)
	};
	}

	function send(entry) {
	if ((window.event.keyCode == 10 \|\| window.event.keyCode == 13) && window.event.ctrlKey) {
	try {
	conn.socket.send(JSON.stringify({kind: 'MESSAGE', channel: channel, message: entry.value}));
	entry.value = '';
	} catch (error) {
	console.log(error);
	}
	return false;
	}
	}

	function join(select) {
	conn.socket.send(JSON.stringify({kind: 'SUBSCRIBE', channel: select.value}));
	}

	// Establish connection
	connect();

	// Dropdown menu initialization
	$('#channel').dropdown();
	</script>
	</body>
	</html>
	<?php

	use io\redis\RedisProtocol;
	use websocket\Listeners;

	class Chat extends Listeners {
	private $subscriptions= [];

	/** @return [:var] */
	public function serve($events) {
	$dsn= $this->environment->arguments()[0] ?? 'redis://localhost';

	// Subscribe to Redis queues
	$sub= new RedisProtocol($dsn)->connect();
	$events->add($sub->socket(), fn() => {
	[$type, $channel, $message]= $sub->receive();
	foreach ($this->subscriptions[$channel] ?? [] as $index => $connection) {
	try {
	$connection->send(json_encode(['kind' => 'MESSAGE', 'value' => $message]));
	} catch ($e) {
	unset($this->subscriptions[$channel][$index]); // Client disconnected
	}
	}
	});

	// Handle websocket messages
	$pub= new RedisProtocol($dsn)->connect();
	return [
	'/chat' => fn($connection, $message) => {
	$value= json_decode($message, true);
	$return= switch ($value['kind']) {
	case 'PING' => ['kind' => 'PONG'];

	case 'MESSAGE' => {
	$pub->command('PUBLISH', $value['channel'], $value['message']);
	return ['kind' => 'ACCEPTED'];
	}

	case 'SUBSCRIBE' => {
	foreach ($this->subscriptions as $channel => &$subscriptions) {
	unset($subscriptions[$connection->id()]);
	empty($subscriptions) && $sub->command('UNSUBSCRIBE', $channel);
	}
	$channel= $value['channel'];
	if (!isset($this->subscriptions[$channel])) {
	$sub->command('SUBSCRIBE', $channel);
	$this->subscriptions[$channel]= [];
	}
	$this->subscriptions[$channel][$connection->id()]= $connection;
	return ['kind' => 'JOINED', 'channel' => $channel];
	}

	default => ['kind' => 'ERROR', 'value' => 'Unknown '.$value['kind']];
	};

	$connection->send(json_encode($return));
	return $return['kind'];
	}
	];
	}
	}