A broadcasting

This commit is contained in:
a6a2f5842 2025-09-13 19:33:29 +02:00
parent 5de4d76e91
commit 5c7d6a1c1b
3 changed files with 91 additions and 2 deletions

View File

@ -268,6 +268,36 @@ class Controller
return true; return true;
} }
final public function broadcasting(
array|string|null $payload = null,
?string $event = null,
?string $channel = null,
bool $including_self = false
){
if (is_string($payload)) {
$payload = [
'message' => $payload,
];
}
$p = [
'event' => ($event ?? $this->event),
'data' => $payload,
'channel' => $channel ?? ($this->channel ? $this->channel->getName() : null),
];
if (get_class($this->connection) !== MockConnection::class) {
throw new \Exception('This method is only available in async mode');
}
$connection = clone $this->connection;
$connection->broadcast(
$p,
$channel,
$including_self
);
}
protected static function get_uniquifyer($event) protected static function get_uniquifyer($event)
{ {
preg_match('/[\[].*[\]]/', $event, $matches); preg_match('/[\[].*[\]]/', $event, $matches);

View File

@ -498,7 +498,7 @@ class Handler implements MessageComponentInterface
$connection->send(json_encode([ $connection->send(json_encode([
'event' => $message['event'].':error', 'event' => $message['event'].':error',
'data' => [ 'data' => [
'message' => 'Timeout', 'message' => $message['event'] . ' timeout',
'diff' => $diff, 'diff' => $diff,
], ],
])); ]));
@ -519,8 +519,23 @@ class Handler implements MessageComponentInterface
// Retrieve cached data // Retrieve cached data
$sending = @cache()->get($pidcache_data); $sending = @cache()->get($pidcache_data);
// Send the data to client // Send the data to client
$connection->send($sending); if(@$message['broadcast']){
$bm = json_decode($sending, true);
$this->broadcast(
$connection->app->id,
$bm['data'] ?? null,
$bm['event'] ?? null,
$bm['channel'] ?? null,
$bm['including_self'],
$connection
);
} else{
$connection->send($sending);
}
// Stop periodic check // Stop periodic check
$this->channelManager->loop->cancelTimer($timer); $this->channelManager->loop->cancelTimer($timer);
@ -530,4 +545,35 @@ class Handler implements MessageComponentInterface
pcntl_waitpid(-1, $status, WNOHANG); pcntl_waitpid(-1, $status, WNOHANG);
}); });
} }
public function broadcast(
string $appId,
mixed $payload,
?string $event = null,
?string $channel = null,
bool $including_self = false,
$connection = null
) : void {
$channel = $this->channelManager->findOrCreate($appId,$channel);
foreach ($channel->getConnections() as $channel_conection) {
if ($channel_conection !== $connection) {
$channel_conection->send(json_encode([
'event' => ($event ?? $event),
'data' => $payload,
'channel' => $channel ?? $channel->getName(),
]));
}
if ($including_self) {
$connection->send(json_encode([
'event' => ($event ?? $event),
'data' => $payload,
'channel' => $channel ?? $channel->getName(),
]));
}
}
}
} }

View File

@ -69,6 +69,19 @@ class MockConnection extends Connection implements \Ratchet\ConnectionInterface
return $this; return $this;
} }
public function broadcast(
$data,
?string $channel = null,
bool $including_self = false,
){
$data ??= [];
$data['broadcast'] = true;
$data['channel'] ??= $channel;
$data['including_self'] = $including_self;
return $this->send(json_encode($data));
}
private static function getDataKey() private static function getDataKey()
{ {
$key = 'dedicated_data_' . getmypid(); $key = 'dedicated_data_' . getmypid();