sevenhill/Qlyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(transport): Zstd-распаковка, гонка PacketReceiver, fix имён и стикеров

Browse Source
This commit is contained in:
klockky
2026-05-14 13:12:41 +03:00
parent ea3042f341
commit d541e9df9d
10 changed files with 137 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/backend/api.dart
+9 -1
View File
@@ -266,7 +266,15 @@ class Api {
}
Future<void> _onDataReceived(Uint8List data) async {
await for (final packet in _receiver.feed(data)) {
final rawPackets = _receiver.feed(data);
for (final raw in rawPackets) {
final Packet packet;
try {
packet = await unpackPacket(raw);
} catch (e) {
logger.e('PacketReceiver: ошибка распаковки: $e');
continue;
}
if (packet.isError &&
packet.payload is Map &&
(packet.payload['message'] == 'FAIL_LOGIN_TOKEN' ||
lib/backend/modules/chats.dart
-1
View File
@@ -160,7 +160,6 @@ class ChatsModule {
static Future<List<CachedChat>> getChats(int accountId) async {
try {
final rows = await AppDatabase.loadChats(accountId);
return rows.map(CachedChat.fromDbRow).toList();
} catch (e) {
logger.e("Ошибка при получении чатов: $e");
lib/backend/modules/contacts.dart
+49 -5
View File
@@ -1,4 +1,5 @@
import '../../core/storage/app_database.dart';
import 'messages.dart';
class CachedContact {
final int id;
@@ -44,22 +45,65 @@ class ContactsModule {
final contacts = data['contacts'];
if (contacts is! List || contacts.isEmpty) return;
final rows = contacts
.whereType<Map>()
.map((c) => _parseContact(c.cast<dynamic, dynamic>(), accountId))
.whereType<Map<String, dynamic>>()
.toList();
final rows = <Map<String, dynamic>>[];
for (final raw in contacts.whereType<Map>()) {
final contact = raw.cast<dynamic, dynamic>();
final row = _parseContact(contact, accountId);
if (row != null) rows.add(row);
_primeContactCache(contact);
}
if (rows.isNotEmpty) {
await AppDatabase.saveContacts(rows);
}
}
static void _primeContactCache(Map<dynamic, dynamic> contact) {
final id = contact['id'];
if (id is! int) return;
final names = contact['names'];
if (names is List && names.isNotEmpty) {
final nameRaw = names.firstWhere(
(n) => n is Map && n['type'] == 'ONEME',
orElse: () => names.firstWhere((n) => n is Map, orElse: () => null),
);
if (nameRaw is Map) {
final firstName = (nameRaw['firstName'] as String?) ?? '';
final lastName = nameRaw['lastName'] as String?;
final fullName = (lastName != null && lastName.isNotEmpty)
? '$firstName $lastName'
: firstName;
if (fullName.isNotEmpty) ContactCache.put(id, fullName);
}
}
final baseUrl = contact['baseUrl'] as String?;
if (baseUrl != null && baseUrl.isNotEmpty) {
ContactCache.putAvatar(id, baseUrl);
}
}
static Future<List<CachedContact>> getContacts(int accountId) async {
final rows = await AppDatabase.loadContacts(accountId);
return rows.map(CachedContact.fromDbRow).toList();
}
/// Прогревает in-memory ContactCache из локальных контактов.
/// Нужно вызывать на cold start: иначе кэш пуст до следующего логина.
static Future<void> primeCacheFromDb(int accountId) async {
final contacts = await getContacts(accountId);
for (final c in contacts) {
final fullName = (c.lastName != null && c.lastName!.isNotEmpty)
? '${c.firstName} ${c.lastName}'
: c.firstName;
if (fullName.isNotEmpty) ContactCache.put(c.id, fullName);
if (c.baseUrl != null && c.baseUrl!.isNotEmpty) {
ContactCache.putAvatar(c.id, c.baseUrl);
}
}
}
static Map<String, dynamic>? _parseContact(
Map<dynamic, dynamic> contact,
int accountId,
lib/core/protocol/packet.dart
+40 -15
View File
@@ -1,6 +1,7 @@
import 'dart:typed_data';
import 'dart:isolate';
import 'package:dart_lz4/dart_lz4.dart';
import 'package:es_compression/zstd.dart';
import 'package:msgpack_dart/msgpack_dart.dart' as msgpack;
/// ver(1) + cmd(1) + seq(2) + opcode(2) + packedLen(4) = 10
@@ -129,21 +130,7 @@ Future<Packet> unpackPacket(Uint8List packet) async {
if (payloadBytes.isNotEmpty) {
if (compFlag != 0) {
try {
payloadBytes = lz4Decompress(
payloadBytes,
decompressedSize: _maxDecompressedSize,
);
} catch (_) {
try {
payloadBytes = _lz4BlockDecompress(
payloadBytes,
_maxDecompressedSize,
);
} catch (e) {
throw Exception("LZ4 decompression error: $e");
}
}
payloadBytes = _decompressPayload(payloadBytes);
}
try {
@@ -165,6 +152,44 @@ Future<Packet> unpackPacket(Uint8List packet) async {
});
}
/// Определяет формат сжатия по magic-number и распаковывает payload.
/// Сервер может присылать LZ4 block ИЛИ Zstandard в зависимости от ответа.
Uint8List _decompressPayload(Uint8List src) {
// Zstandard: magic 28 B5 2F FD (little-endian)
if (src.length >= 4 &&
src[0] == 0x28 &&
src[1] == 0xB5 &&
src[2] == 0x2F &&
src[3] == 0xFD) {
try {
final out = zstd.decode(src);
return out is Uint8List ? out : Uint8List.fromList(out);
} catch (e) {
throw Exception('Zstd decompression error: $e');
}
}
// LZ4 frame: magic 04 22 4D 18
if (src.length >= 4 &&
src[0] == 0x04 &&
src[1] == 0x22 &&
src[2] == 0x4D &&
src[3] == 0x18) {
try {
return lz4Decompress(src, decompressedSize: _maxDecompressedSize);
} catch (e) {
throw Exception('LZ4 frame decompression error: $e');
}
}
// По умолчанию — LZ4 block (без magic)
try {
return _lz4BlockDecompress(src, _maxDecompressedSize);
} catch (e) {
throw Exception('LZ4 block decompression error: $e');
}
}
/// LZ4 block декомпрессия (без frame-заголовка).
/// Сервер шлёт именно block-формат, dart_lz4 его не поддерживает.
Uint8List _lz4BlockDecompress(Uint8List src, int maxSize) {
lib/core/transport/receiver.dart
+9 -12
View File
@@ -4,15 +4,16 @@ import '../protocol/packet.dart';
import '../utils/logger.dart';
/// Буфер входящих данных.
/// Копит сырые байты из сокета, собирает из них целые пакеты.
/// Копит сырые байты из сокета, нарезает их на байтовые срезы целых пакетов.
class PacketReceiver {
Uint8List _buffer = Uint8List(0);
static const int _maxBufferSize = 2 * 1024 * 1024; // 2 мегабуйта
/// Добавляет байты в буфер, возвращает поток собранных пакетов.
/// Неполные данные остаются в буфере до следующего вызова.
Stream<Packet> feed(Uint8List data) async* {
/// Добавляет байты в буфер и возвращает все собранные пакеты как сырые срезы.
/// Полностью синхронный — нарезка не блокируется на распаковке, поэтому
/// конкурентные вызовы из stream-листенера не могут пересечься на `_buffer`.
List<Uint8List> feed(Uint8List data) {
final newBuffer = Uint8List(_buffer.length + data.length);
newBuffer.setAll(0, _buffer);
newBuffer.setAll(_buffer.length, data);
@@ -23,9 +24,10 @@ class PacketReceiver {
'PacketReceiver: переполнение буфера (${_buffer.length} B), сброс',
);
reset();
return;
return const [];
}
final packets = <Uint8List>[];
while (_buffer.length >= headerSize) {
final bd = ByteData.view(
_buffer.buffer,
@@ -38,15 +40,10 @@ class PacketReceiver {
if (_buffer.length < totalLength) break;
final packetBytes = Uint8List.sublistView(_buffer, 0, totalLength);
packets.add(Uint8List.sublistView(_buffer, 0, totalLength));
_buffer = _buffer.sublist(totalLength);
try {
yield await unpackPacket(packetBytes);
} catch (e) {
logger.e('PacketReceiver: ошибка распаковки: $e');
}
}
return packets;
}
void reset() {
lib/frontend/widgets/message_bubble.dart
+5 -3
View File
@@ -1270,15 +1270,17 @@ class MessageBubble extends StatelessWidget {
}
Widget _buildStickerAttachment(BuildContext ctx, MessageAttachment sticker) {
final preview = (sticker as dynamic).previewData as String? ?? '';
final url = sticker.baseUrl ?? '';
final preview = sticker.previewData ?? '';
final imageUrl = url.isNotEmpty ? url : preview;
return ClipRRect(
borderRadius: BorderRadius.circular(photoBorderRadius),
child: Stack(
children: [
if (preview.isNotEmpty)
if (imageUrl.isNotEmpty)
Image.network(
preview,
imageUrl,
width: 150,
height: 150,
fit: BoxFit.contain,
lib/main.dart
+5
View File
@@ -7,6 +7,7 @@ import 'package:komet/l10n/app_localizations.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'backend/api.dart';
import 'backend/modules/account.dart';
import 'backend/modules/contacts.dart';
import 'backend/modules/messages.dart';
import 'core/storage/app_database.dart';
import 'core/storage/token_storage.dart';
@@ -36,6 +37,10 @@ Future<Locale> _loadInitialLocale() async {
void main() async {
WidgetsFlutterBinding.ensureInitialized();
await AppDatabase.init();
final activeAccountId = await TokenStorage.getActiveAccountId();
if (activeAccountId != null) {
await ContactsModule.primeCacheFromDb(activeAccountId);
}
await api.connect();
final initialLocale = await _loadInitialLocale();
lib/models/attachment.dart
+3 -3
View File
@@ -304,9 +304,9 @@ class StickerAttachment extends MessageAttachment {
return StickerAttachment(
previewData: previewStr,
baseUrl: map['baseUrl'] as String?,
stickerId: map['stickerId'] as String?,
stickerPackId: map['stickerPackId'] as String?,
baseUrl: (map['url'] ?? map['baseUrl'])?.toString(),
stickerId: map['stickerId']?.toString(),
stickerPackId: (map['stickerPackId'] ?? map['setId'])?.toString(),
width: map['width'] as int?,
height: map['height'] as int?,
);
pubspec.lock
+16
View File
@@ -1,6 +1,14 @@
# Generated by pub
# See https://dart.dev/tools/pub/glossary#lockfile
packages:
args:
dependency: transitive
description:
name: args
sha256: d0481093c50b1da8910eb0bb301626d4d8eb7284aa739614d2b394ee09e3ea04
url: "https://pub.dev"
source: hosted
version: "2.7.0"
async:
dependency: transitive
description:
@@ -105,6 +113,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.0.8"
es_compression:
dependency: "direct main"
description:
name: es_compression
sha256: c1ff7af54802631cf5c3942cb67bb99daadcc087f573ca99a9de91002d1a7ece
url: "https://pub.dev"
source: hosted
version: "2.0.15"
fake_async:
dependency: transitive
description:
pubspec.yaml
+1
View File
@@ -38,6 +38,7 @@ dependencies:
# Use with the CupertinoIcons class for iOS style icons.
cupertino_icons: ^1.0.8
dart_lz4: ^1.0.0
es_compression: ^2.0.15
msgpack_dart: ^1.0.1
logger: ^2.6.2
device_info_plus: 12.3.0