added wss tools 6

This commit is contained in:
2026-06-18 19:32:28 +03:30
parent dc440abf61
commit 017ab348c7
12 changed files with 133 additions and 90 deletions
+2
View File
@@ -40,6 +40,8 @@ class LivePriceCacheService
app(ProductDisplayPriceService::class)->cacheLivePricePayload($body);
app(PriceBroadcastService::class)->broadcastLastIfChanged($body);
return $body;
}
+2 -1
View File
@@ -84,7 +84,8 @@ class LivePriceTickService
app(LivePriceCacheService::class)->markTick();
app(PriceBroadcastService::class)->broadcastAfterPriceUpdate($payload);
// last از put داخل lastPrice؛ list بعد از sync محصولات
app(PriceBroadcastService::class)->broadcastListIfChanged();
} finally {
$lock->release();
}
+44 -6
View File
@@ -45,17 +45,55 @@ class PriceBroadcastService
$lastChanged = is_array($lastPayload) && $lastPayload !== []
&& $this->rememberHashIfChanged(self::LAST_HASH_KEY, $lastPayload);
$listPayload = $this->buildProductListPayload();
$listChanged = $listPayload !== null
&& $this->rememberHashIfChanged(self::LIST_HASH_KEY, $listPayload);
$this->publishListIfChanged();
if ($lastChanged) {
if ($lastChanged && is_array($lastPayload)) {
$stream->publish('last', $lastPayload);
}
}
if ($listChanged) {
$stream->publish('list', $listPayload);
public function broadcastLastIfChanged(?array $lastPayload): void
{
$stream = app(PriceStreamService::class);
if (! $stream->isEnabled()) {
return;
}
if (! is_array($lastPayload) || $lastPayload === []) {
return;
}
if (! $this->rememberHashIfChanged(self::LAST_HASH_KEY, $lastPayload)) {
return;
}
$stream->publish('last', $lastPayload);
}
public function broadcastListIfChanged(): bool
{
$stream = app(PriceStreamService::class);
if (! $stream->isEnabled()) {
return false;
}
return $this->publishListIfChanged();
}
private function publishListIfChanged(): bool
{
$listPayload = $this->buildProductListPayload();
if ($listPayload === null) {
return false;
}
if (! $this->rememberHashIfChanged(self::LIST_HASH_KEY, $listPayload)) {
return false;
}
app(PriceStreamService::class)->publish('list', $listPayload);
return true;
}
/**
+8 -13
View File
@@ -68,9 +68,9 @@ class PriceStreamService
return '';
}
public function currentSeq(string $host): int
public function currentSeq(): int
{
$inbox = Cache::get($this->inboxKey($host), ['seq' => 0, 'events' => []]);
$inbox = Cache::get($this->inboxKey(), ['seq' => 0, 'events' => []]);
return (int) ($inbox['seq'] ?? 0);
}
@@ -78,9 +78,9 @@ class PriceStreamService
/**
* @return array<int, array{id: int, event: string, data: mixed}>
*/
public function consumeSince(string $host, int $afterId): array
public function consumeSince(int $afterId): array
{
$inbox = Cache::get($this->inboxKey($host), ['seq' => 0, 'events' => []]);
$inbox = Cache::get($this->inboxKey(), ['seq' => 0, 'events' => []]);
$events = is_array($inbox['events'] ?? null) ? $inbox['events'] : [];
return array_values(array_filter(
@@ -92,18 +92,13 @@ class PriceStreamService
/**
* @param array<string, mixed>|array<int, mixed> $data
*/
public function publish(string $event, array $data, ?string $host = null): void
public function publish(string $event, array $data): void
{
if (! $this->isEnabled()) {
return;
}
$host = $host ?? $this->resolveHost();
if ($host === '') {
return;
}
$key = $this->inboxKey($host);
$key = $this->inboxKey();
$inbox = Cache::get($key, ['seq' => 0, 'events' => []]);
$inbox['seq'] = (int) ($inbox['seq'] ?? 0) + 1;
$inbox['events'][] = [
@@ -119,9 +114,9 @@ class PriceStreamService
Cache::put($key, $inbox, now()->addSeconds(self::INBOX_TTL_SECONDS));
}
private function inboxKey(string $host): string
private function inboxKey(): string
{
return self::INBOX_PREFIX.$this->normalizeHost($host);
return self::INBOX_PREFIX.'global';
}
private function normalizeHost(string $host): string
+8
View File
@@ -77,6 +77,10 @@ class ProductPriceSyncService
app(PriceSyncLogService::class)->record($source, $sourceUrl, $liquidGoldPrice, $logItems);
if ($synced) {
app(PriceBroadcastService::class)->broadcastListIfChanged();
}
return $synced;
}
@@ -101,6 +105,10 @@ class ProductPriceSyncService
$synced = true;
}
if ($synced) {
app(PriceBroadcastService::class)->broadcastListIfChanged();
}
return $synced;
}