thekid/Chat.class.php

## Chat.class.php
<?php namespace de\thekid\example;

use io\redis\RedisProtocol;
use peer\SocketException;
use web\protocol\{Protocols, Http, WebSockets};
use web\{Service, Listeners};
use xp\web\ServeDocumentRootStatically;

class Chat extends Service {
  private $connections= [];

  public function serve($server, $environment) {
    $dsn= $environment->arguments()[0] ?? 'redis://localhost';

    // Subscribe to Redis queues
    $sub= new RedisProtocol($dsn);
    $sub->command('SUBSCRIBE', ':broadcast');
    $server->select($sub->socket(), function() use($sub) {
      [$type, $channel, $message]= $sub->receive();
      foreach ($this->connections[$channel] ?? [] as $index => $connection) {
        try {
          $connection->send(json_encode(['kind' => 'MESSAGE', 'value' => $message]));
        } catch (SocketException $e) {
          unset($this->connections[$channel][$index]);  // Client disconnected
        }
      }
    });

    // Publish to Redis queues from the listener
    $pub= new RedisProtocol($dsn);
    $listener= newinstance(Listeners::class, [$environment], [
      'connections' => null,
      'on' => function() use($pub, $sub) {
        return ['/chat' => function($connection, $message) use($pub, $sub) {
          $value= json_decode($message, true);
          switch ($value['kind']) {
            case 'MESSAGE':
              $r= $pub->command('PUBLISH', $value['value']['channel'], $value['value']['message']);
              $return= ['kind' => 'ACCEPTED'];
              break;

            case 'PING';
              $return= ['kind' => 'PONG'];
              break;

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

            default:
              $return= ['kind' => 'ERROR', 'value' => 'Unknown '.$value['kind']];
          }
          $connection->send(json_encode($return));
          return $return['kind'];
        }];
      }
    ]);

    $listener->connections= &$this->connections;
    $logging= $environment->logging();
    return new Protocols(
      new Http(new ServeDocumentRootStatically($environment), $logging),
      ['websocket' => new WebSockets($listener, $logging)]
    );
  }
}

## 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/websocket");
        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 'CHANNEL': add('Channel', 'You are now on ' + message.value); channel = message.value; 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', value: {channel: channel, message: entry.value}}));
            entry.value = '';
          } catch (error) {
            console.log(error);
          }
          return false;
        }
      }

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

      // Establish connection
      connect();

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

	use io\redis\RedisProtocol;
	use peer\SocketException;
	use web\protocol\{Protocols, Http, WebSockets};
	use web\{Service, Listeners};
	use xp\web\ServeDocumentRootStatically;

	class Chat extends Service {
	private $connections= [];

	public function serve($server, $environment) {
	$dsn= $environment->arguments()[0] ?? 'redis://localhost';

	// Subscribe to Redis queues
	$sub= new RedisProtocol($dsn);
	$sub->command('SUBSCRIBE', ':broadcast');
	$server->select($sub->socket(), function() use($sub) {
	[$type, $channel, $message]= $sub->receive();
	foreach ($this->connections[$channel] ?? [] as $index => $connection) {
	try {
	$connection->send(json_encode(['kind' => 'MESSAGE', 'value' => $message]));
	} catch (SocketException $e) {
	unset($this->connections[$channel][$index]); // Client disconnected
	}
	}
	});

	// Publish to Redis queues from the listener
	$pub= new RedisProtocol($dsn);
	$listener= newinstance(Listeners::class, [$environment], [
	'connections' => null,
	'on' => function() use($pub, $sub) {
	return ['/chat' => function($connection, $message) use($pub, $sub) {
	$value= json_decode($message, true);
	switch ($value['kind']) {
	case 'MESSAGE':
	$r= $pub->command('PUBLISH', $value['value']['channel'], $value['value']['message']);
	$return= ['kind' => 'ACCEPTED'];
	break;

	case 'PING';
	$return= ['kind' => 'PONG'];
	break;

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

	default:
	$return= ['kind' => 'ERROR', 'value' => 'Unknown '.$value['kind']];
	}
	$connection->send(json_encode($return));
	return $return['kind'];
	}];
	}
	]);

	$listener->connections= &$this->connections;
	$logging= $environment->logging();
	return new Protocols(
	new Http(new ServeDocumentRootStatically($environment), $logging),
	['websocket' => new WebSockets($listener, $logging)]
	);
	}
	}
	<!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/websocket");
	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 'CHANNEL': add('Channel', 'You are now on ' + message.value); channel = message.value; 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', value: {channel: channel, message: entry.value}}));
	entry.value = '';
	} catch (error) {
	console.log(error);
	}
	return false;
	}
	}

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

	// Establish connection
	connect();

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