added wss tools 4
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
<?php
|
||||
|
||||
namespace App\Http\Controllers\Site;
|
||||
|
||||
use App\Http\Controllers\Controller;
|
||||
use App\Services\LivePriceCacheService;
|
||||
use App\Services\PriceStreamService;
|
||||
use Illuminate\Http\Request;
|
||||
use Symfony\Component\HttpFoundation\StreamedResponse;
|
||||
|
||||
class PriceStreamController extends Controller
|
||||
{
|
||||
public function stream(Request $request): StreamedResponse
|
||||
{
|
||||
$service = app(PriceStreamService::class);
|
||||
|
||||
if (! $service->isEnabled()) {
|
||||
abort(404);
|
||||
}
|
||||
|
||||
$host = $service->resolveHost($request);
|
||||
|
||||
return response()->stream(function () use ($service, $host) {
|
||||
@set_time_limit(0);
|
||||
|
||||
if (function_exists('apache_setenv')) {
|
||||
@apache_setenv('no-gzip', '1');
|
||||
}
|
||||
|
||||
@ini_set('zlib.output_compression', '0');
|
||||
@ini_set('implicit_flush', '1');
|
||||
|
||||
while (ob_get_level() > 0) {
|
||||
@ob_end_flush();
|
||||
}
|
||||
|
||||
$send = static function (string $event, $data): void {
|
||||
echo 'event: '.$event."\n";
|
||||
echo 'data: '.json_encode($data, JSON_UNESCAPED_UNICODE)."\n\n";
|
||||
@flush();
|
||||
};
|
||||
|
||||
$cache = app(LivePriceCacheService::class);
|
||||
$last = $cache->get() ?? $cache->buildFromDatabase();
|
||||
if (is_array($last) && $last !== []) {
|
||||
$send('last', $last);
|
||||
}
|
||||
|
||||
$lastId = $host !== '' ? $service->currentSeq($host) : 0;
|
||||
$heartbeatTicks = 0;
|
||||
|
||||
while (! connection_aborted()) {
|
||||
if ($host !== '') {
|
||||
foreach ($service->consumeSince($host, $lastId) as $event) {
|
||||
$send((string) $event['event'], $event['data']);
|
||||
$lastId = (int) $event['id'];
|
||||
}
|
||||
}
|
||||
|
||||
$heartbeatTicks++;
|
||||
if ($heartbeatTicks >= 30) {
|
||||
echo ': ping '.time()."\n\n";
|
||||
@flush();
|
||||
$heartbeatTicks = 0;
|
||||
}
|
||||
|
||||
usleep(500000);
|
||||
}
|
||||
}, 200, [
|
||||
'Content-Type' => 'text/event-stream; charset=UTF-8',
|
||||
'Cache-Control' => 'no-cache, no-store, must-revalidate',
|
||||
'Connection' => 'keep-alive',
|
||||
'X-Accel-Buffering' => 'no',
|
||||
]);
|
||||
}
|
||||
}
|
||||
@@ -8,25 +8,19 @@ use Illuminate\Support\Facades\Cache;
|
||||
|
||||
class PriceBroadcastService
|
||||
{
|
||||
private const LAST_HASH_KEY = 'ws_broadcast:last_hash';
|
||||
private const LAST_HASH_KEY = 'price_broadcast:last_hash';
|
||||
|
||||
private const LIST_HASH_KEY = 'ws_broadcast:list_hash';
|
||||
private const LIST_HASH_KEY = 'price_broadcast:list_hash';
|
||||
|
||||
/**
|
||||
* montzar فقط trigger «price_update» را به کلاینتها relay میکند.
|
||||
* برادکست SSE — payload کامل last/list وقتی داده عوض شود.
|
||||
*
|
||||
* @param array<string, mixed>|null $lastPayload
|
||||
*/
|
||||
public function broadcastAfterPriceUpdate(?array $lastPayload = null, bool $force = false): void
|
||||
{
|
||||
$ws = app(WebsocketBroadcastService::class);
|
||||
if (! $ws->isEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
if ($force) {
|
||||
$ws->sendEvent('price_update', []);
|
||||
|
||||
$stream = app(PriceStreamService::class);
|
||||
if (! $stream->isEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -35,6 +29,19 @@ class PriceBroadcastService
|
||||
$lastPayload = $cache->get() ?? $cache->buildFromDatabase();
|
||||
}
|
||||
|
||||
if ($force) {
|
||||
if (is_array($lastPayload) && $lastPayload !== []) {
|
||||
$stream->publish('last', $lastPayload);
|
||||
}
|
||||
|
||||
$listPayload = $this->buildProductListPayload();
|
||||
if ($listPayload !== null) {
|
||||
$stream->publish('list', $listPayload);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$lastChanged = is_array($lastPayload) && $lastPayload !== []
|
||||
&& $this->rememberHashIfChanged(self::LAST_HASH_KEY, $lastPayload);
|
||||
|
||||
@@ -42,8 +49,12 @@ class PriceBroadcastService
|
||||
$listChanged = $listPayload !== null
|
||||
&& $this->rememberHashIfChanged(self::LIST_HASH_KEY, $listPayload);
|
||||
|
||||
if ($lastChanged || $listChanged) {
|
||||
$ws->sendEvent('price_update', []);
|
||||
if ($lastChanged) {
|
||||
$stream->publish('last', $lastPayload);
|
||||
}
|
||||
|
||||
if ($listChanged) {
|
||||
$stream->publish('list', $listPayload);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
<?php
|
||||
|
||||
namespace App\Services;
|
||||
|
||||
use Illuminate\Http\Request;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
|
||||
class PriceStreamService
|
||||
{
|
||||
private const INBOX_PREFIX = 'price_sse_inbox:';
|
||||
|
||||
private const INBOX_TTL_SECONDS = 300;
|
||||
|
||||
private const MAX_EVENTS = 50;
|
||||
|
||||
public function isEnabled(): bool
|
||||
{
|
||||
return app(ConfigService::class)->getBool('wss_enabled', true);
|
||||
}
|
||||
|
||||
public function streamPath(): string
|
||||
{
|
||||
$config = app(ConfigService::class);
|
||||
$path = trim($config->getString('sse_path', ''));
|
||||
|
||||
if ($path === '') {
|
||||
$path = trim($config->getString('wss_path', ''));
|
||||
}
|
||||
|
||||
if ($path === '' || strpos($path, '/prc/') !== false) {
|
||||
return '/api/price/stream';
|
||||
}
|
||||
|
||||
return strpos($path, '/') === 0 ? $path : "/{$path}";
|
||||
}
|
||||
|
||||
public function resolveHost(?Request $request = null): string
|
||||
{
|
||||
$config = app(ConfigService::class);
|
||||
|
||||
$appUrl = trim($config->getString('app_url', ''));
|
||||
if ($appUrl !== '') {
|
||||
$host = parse_url($appUrl, PHP_URL_HOST);
|
||||
if (is_string($host) && $host !== '') {
|
||||
return $this->normalizeHost($host);
|
||||
}
|
||||
}
|
||||
|
||||
if ($request !== null) {
|
||||
$requestHost = trim((string) $request->getHost());
|
||||
if ($requestHost !== '') {
|
||||
return $this->normalizeHost($requestHost);
|
||||
}
|
||||
}
|
||||
|
||||
if (! app()->runningInConsole() && request()) {
|
||||
$requestHost = trim((string) request()->getHost());
|
||||
if ($requestHost !== '') {
|
||||
return $this->normalizeHost($requestHost);
|
||||
}
|
||||
}
|
||||
|
||||
$nameEn = trim((string) $config->getString('name_en', ''));
|
||||
if ($nameEn !== '') {
|
||||
return $this->normalizeHost($nameEn);
|
||||
}
|
||||
|
||||
return '';
|
||||
}
|
||||
|
||||
public function currentSeq(string $host): int
|
||||
{
|
||||
$inbox = Cache::get($this->inboxKey($host), ['seq' => 0, 'events' => []]);
|
||||
|
||||
return (int) ($inbox['seq'] ?? 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array<int, array{id: int, event: string, data: mixed}>
|
||||
*/
|
||||
public function consumeSince(string $host, int $afterId): array
|
||||
{
|
||||
$inbox = Cache::get($this->inboxKey($host), ['seq' => 0, 'events' => []]);
|
||||
$events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];
|
||||
|
||||
return array_values(array_filter(
|
||||
$events,
|
||||
static fn (array $event): bool => (int) ($event['id'] ?? 0) > $afterId
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed>|array<int, mixed> $data
|
||||
*/
|
||||
public function publish(string $event, array $data, ?string $host = null): void
|
||||
{
|
||||
if (! $this->isEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
$host = $host ?? $this->resolveHost();
|
||||
if ($host === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
$key = $this->inboxKey($host);
|
||||
$inbox = Cache::get($key, ['seq' => 0, 'events' => []]);
|
||||
$inbox['seq'] = (int) ($inbox['seq'] ?? 0) + 1;
|
||||
$inbox['events'][] = [
|
||||
'id' => $inbox['seq'],
|
||||
'event' => $event,
|
||||
'data' => $data,
|
||||
];
|
||||
|
||||
if (count($inbox['events']) > self::MAX_EVENTS) {
|
||||
$inbox['events'] = array_slice($inbox['events'], -self::MAX_EVENTS);
|
||||
}
|
||||
|
||||
Cache::put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS));
|
||||
}
|
||||
|
||||
private function inboxKey(string $host): string
|
||||
{
|
||||
return self::INBOX_PREFIX.$this->normalizeHost($host);
|
||||
}
|
||||
|
||||
private function normalizeHost(string $host): string
|
||||
{
|
||||
$host = strtolower(trim($host));
|
||||
|
||||
return strpos($host, 'www.') === 0 ? substr($host, 4) : $host;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user