fix some issue from sse

This commit is contained in:
2026-08-27 11:53:29 +03:30
parent 40b185187c
commit 17d79c0ce6
2 changed files with 95 additions and 19 deletions
@@ -53,33 +53,48 @@ class PriceStreamController extends Controller
$cache = app(LivePriceCacheService::class); $cache = app(LivePriceCacheService::class);
$config = app(\App\Services\ConfigService::class); $config = app(\App\Services\ConfigService::class);
$last = $cache->get(); $specialPricing = app(\App\Services\CustomerSpecialPricingService::class);
// کش ممکن است no_quote صفرِ کوتاه‌عمر داشته باشد؛ با build از montzar_live_price جبران کن $zaarbox = app(ZaarboxPriceService::class);
if (! is_array($last) || ! $config->payloadHasDisplayablePrice($last)) { $broadcast = app(PriceBroadcastService::class);
$last = $cache->buildFromDatabase();
} $resolveLastPayload = static function () use ($cache, $config, $specialPricing, $zaarbox, $customerId): ?array {
if (is_array($last) && $last !== []) { $last = $cache->get();
if ($customerId > 0) { if (! is_array($last) || ! $config->payloadHasDisplayablePrice($last)) {
$last = app(\App\Services\CustomerSpecialPricingService::class) $last = $cache->buildFromDatabase();
->applyToLastPayload($last, $customerId);
} }
$send('last', app(ZaarboxPriceService::class)->attachGoldOunceToPayload($last)); if (! is_array($last) || $last === []) {
return null;
}
if ($customerId > 0) {
$last = $specialPricing->applyToLastPayload($last, $customerId);
}
return $zaarbox->attachGoldOunceToPayload($last);
};
$last = $resolveLastPayload();
$lastSentIdentity = null;
if (is_array($last) && $last !== []) {
$send('last', $last);
$lastSentIdentity = PriceBroadcastService::lastPriceIdentity($last);
} }
$hoursSettings = app(\App\Services\ConfigService::class)->clientTradingHoursSettings(); $hoursSettings = $config->clientTradingHoursSettings();
if ($hoursSettings !== []) { if ($hoursSettings !== []) {
$send('settings', $hoursSettings); $send('settings', $hoursSettings);
} }
$list = app(PriceBroadcastService::class)->buildListPayload(false); $list = $broadcast->buildListPayload(false);
$lastSentListIdentity = null;
if (is_array($list) && $list !== []) { if (is_array($list) && $list !== []) {
$send('list', $list); $send('list', $list);
$lastSentListIdentity = PriceBroadcastService::listIdentity($list);
} }
if ($customerId > 0) { if ($customerId > 0) {
$send( $send(
'special-pricing', 'special-pricing',
app(\App\Services\CustomerSpecialPricingService::class)->clientPayloadForCustomer($customerId) $specialPricing->clientPayloadForCustomer($customerId)
); );
} }
@@ -109,8 +124,16 @@ class PriceStreamController extends Controller
while (! connection_aborted()) { while (! connection_aborted()) {
foreach ($service->consumeSince($lastId) as $event) { foreach ($service->consumeSince($lastId) as $event) {
$send((string) $event['event'], $event['data']); $eventName = (string) $event['event'];
$eventData = $event['data'];
$send($eventName, $eventData);
$lastId = (int) $event['id']; $lastId = (int) $event['id'];
if ($eventName === 'last' && is_array($eventData)) {
$lastSentIdentity = PriceBroadcastService::lastPriceIdentity($eventData);
}
if ($eventName === 'list' && is_array($eventData)) {
$lastSentListIdentity = PriceBroadcastService::listIdentity($eventData);
}
} }
if ($customerId > 0) { if ($customerId > 0) {
@@ -121,12 +144,36 @@ class PriceStreamController extends Controller
} }
$heartbeatTicks++; $heartbeatTicks++;
// هر ۱ ثانیه: قیمت را مستقیم از کش/DB بکش
// (اگر inbox بین tick و PHP-FPM هم‌رسانی نشود، فقط ping دیده می‌شد)
if ($heartbeatTicks % 2 === 0) {
$pulled = $resolveLastPayload();
if (is_array($pulled) && $config->payloadHasDisplayablePrice($pulled)) {
$identity = PriceBroadcastService::lastPriceIdentity($pulled);
if ($lastSentIdentity !== $identity) {
$send('last', $pulled);
$lastSentIdentity = $identity;
}
}
}
// هر ۵ ثانیه — LiteSpeed/پروکسی اتصال بیکار را نمی‌بندد // هر ۵ ثانیه — LiteSpeed/پروکسی اتصال بیکار را نمی‌بندد
if ($heartbeatTicks >= 10) { if ($heartbeatTicks >= 10) {
$send('ping', ['ts' => time()]); $send('ping', ['ts' => time()]);
if ($customerId > 0) { if ($customerId > 0) {
$presence->touch($customerId); $presence->touch($customerId);
} }
$pulledList = $broadcast->buildListPayload(true);
if (is_array($pulledList) && $pulledList !== []) {
$listIdentity = PriceBroadcastService::listIdentity($pulledList);
if ($lastSentListIdentity !== $listIdentity) {
$send('list', $pulledList);
$lastSentListIdentity = $listIdentity;
}
}
$heartbeatTicks = 0; $heartbeatTicks = 0;
} }
+34 -5
View File
@@ -2,8 +2,10 @@
namespace App\Services; namespace App\Services;
use Illuminate\Contracts\Cache\Repository;
use Illuminate\Http\Request; use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;
class PriceStreamService class PriceStreamService
{ {
@@ -99,6 +101,31 @@ class PriceStreamService
$this->publishToInbox($this->customerInboxKey($customerId), $event, $data); $this->publishToInbox($this->customerInboxKey($customerId), $event, $data);
} }
/**
* Inbox باید بین tick (CLI) و SSE (PHP-FPM) مشترک باشد.
* file cache روی چند ورکر/سرور inbox را جدا می‌کند → فقط ping دیده می‌شود.
*/
private function store(): Repository
{
static $store = null;
if ($store instanceof Repository) {
return $store;
}
try {
$redis = Cache::store('redis');
$redis->get('price_sse_inbox:ping');
$store = $redis;
} catch (\Throwable $e) {
Log::warning('PriceStreamService: redis unavailable, falling back to default cache', [
'message' => $e->getMessage(),
]);
$store = Cache::store();
}
return $store;
}
private function inboxKey(): string private function inboxKey(): string
{ {
return self::INBOX_PREFIX.'global'; return self::INBOX_PREFIX.'global';
@@ -111,7 +138,7 @@ class PriceStreamService
private function currentInboxSeq(string $key): int private function currentInboxSeq(string $key): int
{ {
$inbox = Cache::get($key, ['seq' => 0, 'events' => []]); $inbox = $this->store()->get($key, ['seq' => 0, 'events' => []]);
return (int) ($inbox['seq'] ?? 0); return (int) ($inbox['seq'] ?? 0);
} }
@@ -121,7 +148,7 @@ class PriceStreamService
*/ */
private function consumeInboxSince(string $key, int $afterId): array private function consumeInboxSince(string $key, int $afterId): array
{ {
$inbox = Cache::get($key, ['seq' => 0, 'events' => []]); $inbox = $this->store()->get($key, ['seq' => 0, 'events' => []]);
$events = is_array($inbox['events'] ?? null) ? $inbox['events'] : []; $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];
return array_values(array_filter( return array_values(array_filter(
@@ -135,8 +162,10 @@ class PriceStreamService
*/ */
private function publishToInbox(string $key, string $event, array $data): void private function publishToInbox(string $key, string $event, array $data): void
{ {
Cache::lock($key.':lock', 5)->block(3, function () use ($key, $event, $data): void { $store = $this->store();
$inbox = Cache::get($key, ['seq' => 0, 'events' => []]);
$store->lock($key.':lock', 5)->block(3, function () use ($store, $key, $event, $data): void {
$inbox = $store->get($key, ['seq' => 0, 'events' => []]);
$events = is_array($inbox['events'] ?? null) ? $inbox['events'] : []; $events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];
$lastEvent = $events !== [] ? $events[array_key_last($events)] : null; $lastEvent = $events !== [] ? $events[array_key_last($events)] : null;
@@ -157,7 +186,7 @@ class PriceStreamService
$inbox['events'] = array_slice($inbox['events'], -self::MAX_EVENTS); $inbox['events'] = array_slice($inbox['events'], -self::MAX_EVENTS);
} }
Cache::put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS)); $store->put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS));
}); });
} }