How to "share" connections between multiple servers

[thorewi]

Hi,

we have 2 servers for high availability and load balancing. I'm using workerman as websocket server.

Device 1 connects to server 1, device 2 connects to server 2. Device 1 sends message to server 1. What is the best way to send it to server 2 and then to device 2?

Thank you for your advice.

[walkor]

Using the channel service component, it can communicate across servers/processes. The process is for the ws1 server to notify the ws2 server using the channel service component, and then the ws2 server sends data to the specified client.

https://manual.workerman.net/doc/en/components/channel-examples.html

[thorewi]

Ok, but it assumes only one Channel\Server... and I don't want to have one centralized service... maybe I can just connect one server to another also by a websocket?

[walkor]

Yes, you can.

[thorewi]

ok and is there a way how to use workerman (or other component from workerman family) as a websocket client? because right now I'm using Ratchet\Pawl, but there is no support for php 8.4 and the whole project seems little bit dead... thank you.

[walkor]

See example for AsyncTcpConnection

[thorewi]

ok, i created a function like this:

protected function connectServers(bool $isRestart = false): void
	{
		if (empty($_ENV['OTHER_INSTANCE'])) {
			return;
		}

		if ($isRestart) {
			sleep(static::ReconnectInterval);
			if ($this->otherInstanceTimer) {
				Timer::del($this->otherInstanceTimer);
			}
			$this->otherInstanceTime = null;
		}

		$connection = new AsyncTcpConnection('ws://' . $_ENV['OTHER_INSTANCE'] . ':10123');
		$this->otherInstanceConnection = $connection;
		$connection->onConnect = function(AsyncTcpConnection $connection) {
			$this->otherInstanceTimer = Timer::add(static::PingInterval, function() use ($connection) {
				if (!$this->otherInstanceTime) {
					$this->otherInstanceTime = time();
				}
				if (time() - $this->otherInstanceTime > static::PingInterval) {
					$connection->close();
				} else {
					$connection->send($this->generateWebSocketPingFrame(), true);
				}
			});
		};
		$connection->onWebSocketPong = function() {
			$this->otherInstanceTime = time();
		};
		$connection->onClose = function () {
			$this->connectServers(isRestart: true);
		};
		$connection->onError = function () {
			$this->connectServers(isRestart: true);
		};
		$connection->connect();
	}

but for sending ping I implemented own mechanism:

protected function generateWebSocketPingFrame(): string
	{
		$opcode = 0x9;
		$payload = "";
		$length = strlen($payload);

		$maskKey = random_bytes(4);

		if ($length < 126) {
			$head = chr(0x80 | $opcode) . chr(0x80 | $length);
		} elseif ($length < 0xFFFF) {
			$head = chr(0x80 | $opcode) . chr(0x80 | 126) . pack("n", $length);
		} else {
			$head = chr(0x80 | $opcode) . chr(0x80 | 127) . pack("N", 0) . pack("N", $length);
		}

		$masked_payload = '';
		for ($i = 0; $i < $length; $i++) {
			$masked_payload .= $payload[$i] ^ $maskKey[$i % 4];
		}

		return $head . $maskKey . $masked_payload;
	}

is there any implementation included in the library? thank you...

[walkor]

The AsyncTcpConnection has its own reconnect() method and does not need to be implemented again.

Change

$connection->onClose = function () {
	$this->connectServers(isRestart: true);
};
$connection->onError = function () {
	$this->connectServers(isRestart: true);
};

to

$connection->onClose = function ($connection) {
	$connection->reconnect(static::ReconnectInterval);
};

[thorewi]

ok and how does it recognize without pingpong that the connection is really live or dead? because the connection can be in "hang" state - client thinks it's still alive but it isn't (because of firewall, server shutdown without TCP RST and so on)...

[walkor]

To timely detect a connection drop, heartbeat detection is essential.

[thorewi]

ok so shouldn't a heartbeat be implemented directly into AsyncTcpConnection with parameter "interval"?

[walkor]

Using a timer and adding an interval parameter in AsyncTcpConnection is not significantly different:

Timer::add(50, function() use ($connection) {
    $connection->send('ping');
});

$connection->setInterval(50, function($connection) {
    $connection->send('ping');
});

Perhaps future versions will consider adding it.