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

feat: migrate networking to Rust core (kolibri) — adapter, media,submodule, CI

Browse Source
This commit is contained in:
klockky
2026-07-11 21:56:48 +03:00
parent 77df71824e
commit faadc95993
18 changed files with 583 additions and 959 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/actions/setup-rust/action.yml
+22
View File
@@ -0,0 +1,22 @@
name: Setup Rust
description: >
Install the Rust toolchain with platform targets and a build cache, so
cargokit can compile the kolibri native library during the Flutter build.
inputs:
targets:
description: Comma-separated rustup targets for the platform being built.
required: true
runs:
using: composite
steps:
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@stable
with:
targets: ${{ inputs.targets }}
- name: Cache Rust build
uses: Swatinem/rust-cache@v2
with:
workspaces: third_party/kolibri/kolibri-dart/rust
.github/workflows/build-android-fcm.yml
+5
View File
@@ -41,6 +41,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Get dependencies
run: flutter pub get
.github/workflows/build-android.yml
+5
View File
@@ -41,6 +41,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Get dependencies
run: flutter pub get
.github/workflows/build-ios.yml
+5
View File
@@ -37,6 +37,11 @@ jobs:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-ios,aarch64-apple-ios-sim,x86_64-apple-ios
- name: Get dependencies
run: flutter pub get
.github/workflows/build-linux.yml
+5
View File
@@ -42,6 +42,11 @@ jobs:
sudo apt-get update
sudo apt-get install -y clang cmake ninja-build pkg-config libgtk-3-dev libsecret-1-dev libjsoncpp-dev libmpv-dev mpv libopus-dev libogg-dev
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-unknown-linux-gnu
- name: Get dependencies
run: flutter pub get
.github/workflows/build-macos.yml
+5
View File
@@ -37,6 +37,11 @@ jobs:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-darwin,x86_64-apple-darwin
- name: Install create-dmg
run: brew install create-dmg
.github/workflows/build-windows.yml
+5
View File
@@ -37,6 +37,11 @@ jobs:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-pc-windows-msvc
- name: Get dependencies
run: flutter pub get
.github/workflows/flutter-dev.yml
+5
View File
@@ -29,6 +29,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Install dependencies
run: flutter pub get
.github/workflows/flutter-main.yml
+21 -4
View File
@@ -28,6 +28,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Install dependencies
run: flutter pub get
@@ -66,7 +71,7 @@ jobs:
- name: Build Android APK
run: flutter build apk --release --flavor komet
web-linux:
linux:
runs-on: ubuntu-latest
steps:
- name: Checkout code
@@ -85,12 +90,14 @@ jobs:
sudo apt-get update
sudo apt-get install -y clang cmake ninja-build pkg-config libgtk-3-dev libsecret-1-dev libjsoncpp-dev
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-unknown-linux-gnu
- name: Install dependencies
run: flutter pub get
- name: Build Web
run: flutter build web --release
- name: Build Linux
run: flutter build linux --release
@@ -108,6 +115,11 @@ jobs:
flutter-version: '3.44.3'
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-pc-windows-msvc
- name: Install dependencies
run: flutter pub get
@@ -128,6 +140,11 @@ jobs:
flutter-version: '3.44.3'
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-ios,aarch64-apple-ios-sim,x86_64-apple-ios,aarch64-apple-darwin,x86_64-apple-darwin
- name: Install dependencies
run: flutter pub get
.github/workflows/release-dev.yml
+21
View File
@@ -42,6 +42,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Get dependencies
run: flutter pub get
@@ -110,6 +115,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-pc-windows-msvc
- name: Get dependencies
run: flutter pub get
- name: Build Windows
@@ -142,6 +151,10 @@ jobs:
run: |
sudo apt-get update
sudo apt-get install -y clang cmake ninja-build pkg-config libgtk-3-dev libsecret-1-dev libjsoncpp-dev libmpv-dev mpv libopus-dev libogg-dev
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-unknown-linux-gnu
- name: Get dependencies
run: flutter pub get
- name: Build Linux
@@ -169,6 +182,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-darwin,x86_64-apple-darwin
- name: Get dependencies
run: flutter pub get
- name: Fix opus/ogg libs for macOS
@@ -199,6 +216,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-ios,aarch64-apple-ios-sim,x86_64-apple-ios
- name: Get dependencies
run: flutter pub get
- name: Build iOS (no codesign)
.github/workflows/release-main.yml
+21
View File
@@ -37,6 +37,11 @@ jobs:
channel: 'stable'
cache: true
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- name: Get dependencies
run: flutter pub get
@@ -105,6 +110,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-pc-windows-msvc
- name: Get dependencies
run: flutter pub get
- name: Build Windows
@@ -137,6 +146,10 @@ jobs:
run: |
sudo apt-get update
sudo apt-get install -y clang cmake ninja-build pkg-config libgtk-3-dev libsecret-1-dev libjsoncpp-dev
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: x86_64-unknown-linux-gnu
- name: Get dependencies
run: flutter pub get
- name: Build Linux
@@ -164,6 +177,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-darwin,x86_64-apple-darwin
- name: Get dependencies
run: flutter pub get
- name: Build macOS
@@ -192,6 +209,10 @@ jobs:
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
channel: 'stable'
- name: Setup Rust
uses: ./.github/actions/setup-rust
with:
targets: aarch64-apple-ios,aarch64-apple-ios-sim,x86_64-apple-ios
- name: Get dependencies
run: flutter pub get
- name: Build iOS (no codesign)
lib/backend/api.dart
+289 -218
View File
@@ -1,46 +1,50 @@
import 'dart:async';
import 'dart:typed_data';
import 'dart:io';
import 'package:device_info_plus/device_info_plus.dart';
import 'package:flutter_timezone/flutter_timezone.dart';
import 'package:kolibri/kolibri.dart';
import 'package:timezone/data/latest_all.dart' as tz;
import '../core/cache/self_presence.dart';
import '../core/config/config.dart';
import '../core/config/countries.dart';
import '../core/config/komet_settings.dart';
import '../core/config/proxy_config.dart';
import '../core/protocol/opcode_map.dart';
import '../core/protocol/packet.dart';
import '../core/storage/device_identity.dart';
import '../core/storage/spoofing_service.dart';
import '../core/transport/connection.dart';
import '../core/transport/dispatcher.dart';
import '../core/transport/receiver.dart';
import '../core/transport/sender.dart';
import '../core/transport/tls_config.dart';
import '../core/transport/traffic_monitor.dart';
import '../core/transport/vpn_bypass.dart';
import '../core/utils/debug_session_log.dart';
import '../core/utils/logger.dart';
import 'package:device_info_plus/device_info_plus.dart';
import 'package:flutter_timezone/flutter_timezone.dart';
import 'package:timezone/data/latest_all.dart' as tz;
import 'dart:io';
enum SessionState { disconnected, connecting, connected, online }
/// Клиент API.
///
/// Подключение, хэндшейк, пинг, реконнект.
/// Тонкий адаптер над Rust-ядром [KolibriSession] (пакет kolibri): подключение,
/// хэндшейк, пинг и реконнект живут в ядре, здесь — оркестрация жизненного цикла
/// и сохранение прежнего интерфейса для модулей (Packet/пуши/стримы).
class Api {
final Connection _connection = Connection();
final PacketReceiver _receiver = PacketReceiver();
final PacketSender _sender = PacketSender();
KolibriSession? _session;
/// Роутер пушей (ответы на запросы ядро матчит само, диспетчер держим только
/// ради registerHandler/pushStream).
final PacketDispatcher _dispatcher = PacketDispatcher();
StreamSubscription<(int, Map<String, dynamic>)>? _pushSub;
SessionState _sessionState = SessionState.disconnected;
final _stateController = StreamController<SessionState>.broadcast();
final _sessionExpiredController =
StreamController<SessionExpiredException>.broadcast();
final _handshakeSuccessController = StreamController<String>.broadcast();
Map<dynamic, dynamic>? _userAgent;
final _errorController = StreamController<String>.broadcast();
Map<dynamic, dynamic>? _userAgent;
Map<dynamic, dynamic>? get userAgent => _userAgent;
int? _callsSeed;
@@ -49,6 +53,9 @@ class Api {
int? get callsSeed => _callsSeed;
String? get deviceId => _deviceId;
/// Сырой доступ к сессии ядра — для медиа-загрузок (data-plane).
KolibriSession? get session => _session;
String? spoofScope;
static bool _tzInitialized = false;
@@ -63,28 +70,31 @@ class Api {
_sessionExpiredController.stream;
Stream<String> get handshakeSuccessStream =>
_handshakeSuccessController.stream;
Stream<String> get errorStream => _dispatcher.errorStream;
Stream<String> get errorStream => _errorController.stream;
SessionState get state => _sessionState;
StreamSubscription<Uint8List>? _dataSubscription;
StreamSubscription<SocketState>? _socketStateSubscription;
Timer? _pingTimer;
Timer? _livenessTimer;
Timer? _reconnectTimer;
Timer? _connectWatchdog;
int _connectGen = 0;
int _reconnectAttempts = 0;
bool _autoReconnect = false;
int _sessionEpoch = 0;
int _seq = 0;
bool? _lastInteractive;
static const Duration _connectWatchdogTimeout = Duration(seconds: 75);
static const Duration _shouldArmTimeout = Duration(seconds: 5);
static const Duration _endpointTimeout = Duration(seconds: 5);
static const Duration _livenessInterval = Duration(seconds: 5);
int get sessionEpoch => _sessionEpoch;
/// Залипает на время сессии: VPN-путь не сработал — идём мимо туннеля.
bool _bypassActive = false;
int _nextSeq() => _seq = (_seq + 1) & 0xFFFF;
// Публичное API
/// Подключается к серверу, шлёт хэндшейк, запускает пинг.
@@ -100,14 +110,6 @@ class Api {
_armConnectWatchdog(gen);
try {
_dataSubscription = _connection.dataStream.listen(_onDataReceived);
_socketStateSubscription = _connection.stateStream.listen((socketState) {
if (socketState == SocketState.disconnected &&
_sessionState != SessionState.disconnected) {
_onDisconnected();
}
});
bool bypassArmed;
try {
bypassArmed = await VpnBypassService.instance.shouldArm().timeout(
@@ -121,9 +123,6 @@ class Api {
if (!bypassArmed) _bypassActive = false;
final useBypass = _bypassActive && bypassArmed;
final attemptTimeout = bypassArmed && !useBypass
? const Duration(seconds: 8)
: null;
({String host, int port}) endpoint;
try {
@@ -137,70 +136,36 @@ class Api {
}
if (gen != _connectGen) return;
final options = await _buildSessionOptions(endpoint);
if (gen != _connectGen) return;
logger.i(
'connect: endpoint ${endpoint.host}:${endpoint.port}, bypass=$useBypass',
);
try {
await _connection.connect(
endpoint.host,
endpoint.port,
bypassVpn: useBypass,
timeout: attemptTimeout,
);
} catch (e) {
if (gen != _connectGen) return;
await _handleConnectFailure(
e,
phase: 'Не удалось подключиться',
bypassArmed: bypassArmed,
useBypass: useBypass,
bypassWhy: 'подключение не удалось',
);
return;
if (useBypass) {
try {
await VpnBypassService.instance.bind();
} catch (e) {
logger.w('connect: VPN bind не удался ($e)');
}
}
if (gen != _connectGen) return;
final session = options;
_session = session;
_pushSub = session.pushesMap().listen(_onPush);
TrafficMonitor.instance.recordEvent(
'connect',
endpoint: '${endpoint.host}:${endpoint.port}',
);
_setSessionState(SessionState.connected);
_reconnectAttempts = 0;
HandshakeInfo info;
try {
logger.i('connect: сокет готов, отправляю хэндшейк');
final response = await sendHandshake();
if (gen != _connectGen) return;
if (response.isOk) {
_callsSeed = response.payload['callsSeed'] as int?;
_registrationCountries = _parseRegistrationCountries(
response.payload,
);
_sessionState = SessionState.online;
_sessionEpoch++;
_cancelConnectWatchdog();
_startPinging();
logger.i('Сессия онлайн, хэндшейк ок');
if (_onReconnectCallback != null) {
try {
await _onReconnectCallback!();
} catch (e) {
logger.w('Авто-логин при хэндшейке не удался: $e');
}
}
if (_sessionState == SessionState.online) {
_stateController.add(SessionState.online);
_handshakeSuccessController.add(
response.payload['device_name'] as String? ?? 'Unknown',
);
}
} else {
logger.e('Хэндшейк отклонён: ${response.payload}');
await _handleConnectFailure(
StateError('хэндшейк отклонён сервером'),
phase: 'Хэндшейк отклонён',
bypassArmed: bypassArmed,
useBypass: useBypass,
bypassWhy: 'хэндшейк отклонён',
disconnectSocket: true,
);
}
info = await session.connect();
} catch (e) {
if (gen != _connectGen) return;
await _handleConnectFailure(
@@ -209,8 +174,34 @@ class Api {
bypassArmed: bypassArmed,
useBypass: useBypass,
bypassWhy: 'хэндшейк не прошёл',
disconnectSocket: true,
);
return;
} finally {
if (useBypass) {
try {
await VpnBypassService.instance.restoreDefault();
} catch (_) {}
}
}
if (gen != _connectGen) return;
_callsSeed = info.callsSeed?.toInt();
_registrationCountries = _parseRegistrationCountries(info.payloadMap);
_sessionState = SessionState.online;
_sessionEpoch++;
_cancelConnectWatchdog();
_startLiveness();
logger.i('Сессия онлайн, хэндшейк ок');
if (_onReconnectCallback != null) {
try {
await _onReconnectCallback!();
} catch (e) {
logger.w('Авто-логин при хэндшейке не удался: $e');
}
}
if (_sessionState == SessionState.online) {
_stateController.add(SessionState.online);
_handshakeSuccessController.add(info.deviceName ?? 'Unknown');
}
} catch (e, st) {
logger.e('connect: непредвиденная ошибка: $e\n$st');
@@ -244,9 +235,6 @@ class Api {
_connectGen++;
_cancelConnectWatchdog();
_cleanup();
try {
await _connection.disconnect();
} catch (_) {}
_setSessionState(SessionState.disconnected);
if (_autoReconnect) _scheduleReconnect();
}
@@ -257,13 +245,11 @@ class Api {
required bool bypassArmed,
required bool useBypass,
required String bypassWhy,
bool disconnectSocket = false,
}) async {
logger.e('$phase: $error');
_cancelConnectWatchdog();
if (_sessionState != SessionState.disconnected) {
_cleanup();
if (disconnectSocket) await _connection.disconnect();
_setSessionState(SessionState.disconnected);
_armBypassIfPossible(bypassArmed, useBypass, bypassWhy);
_scheduleReconnect();
@@ -284,7 +270,6 @@ class Api {
_connectGen++;
_reconnectTimer?.cancel();
_cleanup();
await _connection.disconnect();
_setSessionState(SessionState.disconnected);
}
@@ -303,7 +288,114 @@ class Api {
}
}
Future<Packet> sendHandshake() async {
/// Отправляет запрос и ждёт ответ от сервера.
Future<Packet> sendRequest(int opcode, Map<dynamic, dynamic> payload) async {
final session = _session;
if (session == null) {
throw StateError('Нет соединения (${Opcode.name(opcode)})');
}
final seq = _nextSeq();
DebugSessionLog.instance.recordRequest(opcode, seq, payload);
TrafficMonitor.instance.recordOutgoing(opcode, payload, seq, 0);
final KolibriResponse resp;
try {
resp = await session
.requestMapFull(opcode, Map<String, dynamic>.from(payload))
.timeout(
ServerConfig.requestTimeout,
onTimeout: () =>
throw TimeoutException('${Opcode.name(opcode)} таймаут'),
);
} catch (e, st) {
DebugSessionLog.instance.recordError(seq, e);
Error.throwWithStackTrace(e, st);
}
final packet = Packet(
cmd: resp.cmd,
seq: seq,
opcode: resp.opcode,
payload: resp.payload,
);
DebugSessionLog.instance.recordResponse(seq, packet.cmd, packet.payload);
TrafficMonitor.instance.recordIncoming(packet, 0);
if (packet.isError) {
if (isSessionExpiredPayload(packet.payload)) {
final ex = SessionExpiredException(
messageFromErrorPayload(packet.payload),
);
_sessionExpiredController.add(ex);
DebugSessionLog.instance.recordError(seq, ex);
throw ex;
}
final text = _serverErrorText(packet.payload);
if (text != null) _errorController.add(text);
final err = PacketError(
messageFromErrorPayload(packet.payload),
errorKey: resp.errorKey,
);
DebugSessionLog.instance.recordError(seq, err);
throw err;
}
return packet;
}
Future<Map<dynamic, dynamic>?> sendRequestMap(
int opcode,
Map<dynamic, dynamic> payload,
) async {
final response = await sendRequest(opcode, payload);
if (!response.isOk || response.payload is! Map) return null;
return response.payload as Map<dynamic, dynamic>;
}
Future<bool> sendRequestOk(int opcode, Map<dynamic, dynamic> payload) async {
final response = await sendRequest(opcode, payload);
return response.isOk;
}
Future<Packet> sendRequestOrThrow(
int opcode,
Map<dynamic, dynamic> payload,
) async {
final response = await sendRequest(opcode, payload);
throwIfPacketError(response);
return response;
}
/// Вешает обработчик на пуши с указанным опкодом.
void registerPushHandler(int opcode, void Function(Packet) handler) {
_dispatcher.registerHandler(opcode, handler);
}
/// Снимает обработчик пушей с указанного опкода.
void unregisterPushHandler(int opcode) {
_dispatcher.unregisterHandler(opcode);
}
/// Стрим всех входящих пушей от сервера.
Stream<Packet> get pushStream => _dispatcher.pushStream;
Future<void> dispose() async {
_autoReconnect = false;
_reconnectTimer?.cancel();
_cleanup();
_dispatcher.dispose();
await _stateController.close();
await _sessionExpiredController.close();
await _handshakeSuccessController.close();
await _errorController.close();
}
// Внутрянка
/// Строит устройство-поля и создаёт сессию ядра. Заодно заполняет
/// [_userAgent] и [_deviceId] для геттеров.
Future<KolibriSession> _buildSessionOptions(
({String host, int port}) endpoint,
) async {
final deviceInfo = DeviceInfoPlugin();
String deviceType = 'ANDROID';
@@ -404,95 +496,57 @@ class Api {
'deviceName': deviceName,
'deviceLocale': deviceLocale,
};
_deviceId = deviceId;
final payload = <dynamic, dynamic>{
'mt_instanceid': instanceId,
'userAgent': _userAgent,
'clientSessionId': clientSessionId,
'deviceId': deviceId,
};
final insecureTls = await TlsConfig.isInsecureAllowed();
final proxy = await _buildProxyUrl();
return sendRequest(Opcode.sessionInit, payload);
return openSession(
host: endpoint.host,
port: endpoint.port,
deviceId: deviceId,
instanceId: instanceId,
appVersion: appVersion,
buildNumber: buildNumber,
deviceType: deviceType,
osVersion: osVersion,
timezone: timezone,
screen: screen,
pushDeviceType: pushDeviceType,
arch: architecture,
locale: locale,
deviceName: deviceName,
deviceLocale: deviceLocale,
clientSessionId: clientSessionId,
pingIntervalSecs: ServerConfig.pingInterval.inSeconds,
pingInteractive: !KometSettings.ghostMode.value,
autoReconnect: false,
insecureTls: insecureTls,
proxy: proxy,
);
}
/// Отправляет запрос и ждёт ответ от сервера.
Future<Packet> sendRequest(int opcode, Map<dynamic, dynamic> payload) {
final seq = _sender.send(_connection, opcode, payload);
DebugSessionLog.instance.recordRequest(opcode, seq, payload);
return _dispatcher
.registerPending(seq)
.timeout(
ServerConfig.requestTimeout,
onTimeout: () =>
throw TimeoutException('${Opcode.name(opcode)} таймаут'),
)
.then(
(packet) {
DebugSessionLog.instance.recordResponse(
seq,
packet.cmd,
packet.payload,
);
return packet;
},
onError: (Object e, StackTrace st) {
DebugSessionLog.instance.recordError(seq, e);
Error.throwWithStackTrace(e, st);
},
);
static Future<String?> _buildProxyUrl() async {
final p = await ProxyConfig.load();
if (!p.isEnabled) return null;
final scheme = p.type == ProxyType.socks5 ? 'socks5h' : 'http';
final auth = p.hasCredentials
? '${Uri.encodeComponent(p.username!)}:'
'${Uri.encodeComponent(p.password!)}@'
: '';
return '$scheme://$auth${p.host}:${p.port}';
}
Future<Map<dynamic, dynamic>?> sendRequestMap(
int opcode,
Map<dynamic, dynamic> payload,
) async {
final response = await sendRequest(opcode, payload);
if (!response.isOk || response.payload is! Map) return null;
return response.payload as Map<dynamic, dynamic>;
void _onPush((int, Map<String, dynamic>) event) {
final packet = Packet(
cmd: CmdType.push,
opcode: event.$1,
payload: event.$2,
);
TrafficMonitor.instance.recordIncoming(packet, 0);
_dispatcher.dispatch(packet);
}
Future<bool> sendRequestOk(int opcode, Map<dynamic, dynamic> payload) async {
final response = await sendRequest(opcode, payload);
return response.isOk;
}
Future<Packet> sendRequestOrThrow(
int opcode,
Map<dynamic, dynamic> payload,
) async {
final response = await sendRequest(opcode, payload);
throwIfPacketError(response);
return response;
}
/// Вешает обработчик на пуши с указанным опкодом.
void registerPushHandler(int opcode, void Function(Packet) handler) {
_dispatcher.registerHandler(opcode, handler);
}
/// Снимает обработчик пушей с указанного опкода.
void unregisterPushHandler(int opcode) {
_dispatcher.unregisterHandler(opcode);
}
/// Стрим всех входящих пушей от сервера.
Stream<Packet> get pushStream => _dispatcher.pushStream;
Future<void> dispose() async {
_autoReconnect = false;
_reconnectTimer?.cancel();
_cleanup();
_dispatcher.dispose();
await _connection.dispose();
await _stateController.close();
await _sessionExpiredController.close();
await _handshakeSuccessController.close();
}
// Внутрянка
void _setSessionState(SessionState state) {
if (_sessionState == state) return;
_sessionState = state;
@@ -500,35 +554,6 @@ class Api {
logger.i('Сессия: ${state.name}');
}
Future<void> _onDataReceived(Uint8List data) async {
final List<Uint8List> rawPackets;
try {
rawPackets = _receiver.feed(data);
} on ReceiverOverflowException catch (e) {
logger.e('$e — форсируем реконнект');
if (_sessionState != SessionState.disconnected) {
unawaited(_forceReconnect());
}
return;
}
for (final raw in rawPackets) {
final Packet packet;
try {
packet = await unpackPacket(raw);
} catch (e) {
logger.e('PacketReceiver: ошибка распаковки: $e');
continue;
}
TrafficMonitor.instance.recordIncoming(packet, raw.length);
if (packet.isError && isSessionExpiredPayload(packet.payload)) {
_sessionExpiredController.add(
SessionExpiredException(messageFromErrorPayload(packet.payload)),
);
}
_dispatcher.dispatch(packet);
}
}
void _onDisconnected() {
_connectGen++;
_cleanup();
@@ -536,13 +561,18 @@ class Api {
if (_autoReconnect) _scheduleReconnect();
}
/// Пробный запрос-пинг: если не ответил — форсируем реконнект.
Future<void> _probeLiveness() async {
if (_sessionState != SessionState.online) return;
final session = _session;
if (session == null) return;
final epoch = _sessionEpoch;
try {
await sendRequest(Opcode.ping, {
'interactive': !KometSettings.ghostMode.value,
}).timeout(const Duration(seconds: 6));
await session
.requestMapFull(Opcode.ping, {
'interactive': !KometSettings.ghostMode.value,
})
.timeout(const Duration(seconds: 6));
} catch (_) {
if (_sessionEpoch != epoch || _sessionState != SessionState.online) {
return;
@@ -555,7 +585,6 @@ class Api {
Future<void> _forceReconnect() async {
_connectGen++;
_cleanup();
await _connection.disconnect();
_reconnectAttempts = 0;
_reconnectTimer?.cancel();
_setSessionState(SessionState.disconnected);
@@ -564,12 +593,18 @@ class Api {
void _cleanup() {
_cancelConnectWatchdog();
_pingTimer?.cancel();
_dataSubscription?.cancel();
_socketStateSubscription?.cancel();
_dataSubscription = null;
_socketStateSubscription = null;
_receiver.reset();
_livenessTimer?.cancel();
_livenessTimer = null;
_pushSub?.cancel();
_pushSub = null;
_lastInteractive = null;
final session = _session;
_session = null;
if (session != null) {
try {
session.disconnect();
} catch (_) {}
}
_dispatcher.clearPending();
_handshakeSuccessController.add('disconnected');
}
@@ -584,17 +619,44 @@ class Api {
_onReconnectCallback = callback;
}
void _startPinging() {
_pingTimer?.cancel();
sendPing(interactive: !KometSettings.ghostMode.value);
_pingTimer = Timer.periodic(ServerConfig.pingInterval, (_) {
sendPing(interactive: !KometSettings.ghostMode.value);
});
/// Поллит состояние ядра (стрима состояний нет) — детект разрыва, плюс
/// синхронизация interactive-флага пинга и присутствия.
void _startLiveness() {
_livenessTimer?.cancel();
_lastInteractive = !KometSettings.ghostMode.value;
_livenessTimer = Timer.periodic(_livenessInterval, (_) => _tickLiveness());
}
void _tickLiveness() {
final session = _session;
if (session == null || _sessionState != SessionState.online) return;
final st = session.state();
if (st != 'online' && st != 'connected') {
logger.w('kolibri сессия "$st" — реконнект');
_onDisconnected();
return;
}
final interactive = !KometSettings.ghostMode.value;
if (interactive != _lastInteractive) {
_lastInteractive = interactive;
try {
session.setPingInteractive(interactive: interactive);
} catch (_) {}
}
if (interactive) {
SelfPresence.markOnline();
} else {
SelfPresence.markOfflineFromPing();
}
}
void sendPing({required bool interactive}) {
if (_connection.isConnected) {
_sender.send(_connection, Opcode.ping, {'interactive': interactive});
final session = _session;
if (session != null && _sessionState == SessionState.online) {
try {
session.setPingInteractive(interactive: interactive);
} catch (_) {}
_lastInteractive = interactive;
if (interactive) {
SelfPresence.markOnline();
} else {
@@ -608,6 +670,15 @@ class Api {
return v.substring(v.indexOf('_') + 1, v.length - 1);
}
static String? _serverErrorText(dynamic payload) {
if (payload is! Map) return null;
for (final key in ['localizedMessage', 'title']) {
final v = payload[key];
if (v is String && v.trim().isNotEmpty) return v.trim();
}
return null;
}
static List<CountryName>? _parseRegistrationCountries(dynamic payload) {
if (payload is! Map) return null;
final raw = payload['reg-country-code'];
lib/backend/modules/file_uploader.dart
+120 -566
View File
@@ -3,11 +3,10 @@ import 'dart:convert' show jsonDecode, utf8;
import 'dart:io';
import 'dart:typed_data';
import 'package:kolibri/kolibri.dart' as kb;
import '../api.dart';
import '../../core/config/proxy_config.dart';
import '../../core/protocol/opcode_map.dart';
import '../../core/transport/proxy_connector.dart';
import '../../core/transport/tls_config.dart';
import '../../core/utils/logger.dart';
import 'messages.dart';
@@ -41,10 +40,10 @@ class UploadError extends UploadEvent {
const UploadError(this.message);
}
/// Оркестратор медиа-загрузок: control-plane (URL, отправка сообщения) идёт
/// обычными опкодами, data-plane (заливка на CDN) — через Rust-ядро kolibri,
/// которое стримит файл с диска (не держит его целиком в памяти).
class FileUploader {
static const String _userAgentHeader =
'OKMessages/26.14.1 (Android 11; TECNO MOBILE LIMITED TECNO LE7n; xxhdpi 480dpi 1080x2208)';
final Api api;
final MessagesModule messages;
@@ -62,13 +61,11 @@ class FileUploader {
}) {
final ctrl = StreamController<UploadEvent>();
var cancelled = false;
Socket? socket;
StreamSubscription<kb.UploadEvent>? sub;
ctrl.onCancel = () {
cancelled = true;
try {
socket?.destroy();
} catch (_) {}
sub?.cancel();
};
Future<void> run() async {
@@ -80,6 +77,12 @@ class FileUploader {
return;
}
final session = api.session;
if (session == null) {
ctrl.add(const UploadError('no_session'));
return;
}
unawaited(() async {
try {
await api.sendRequest(Opcode.msgTyping, {
@@ -89,27 +92,49 @@ class FileUploader {
} catch (_) {}
}());
final uri = Uri.parse(info.url);
final result = await _sendHttpRequest(
uri,
method: 'POST',
headers: _buildUploadHeaders(uri, filename, totalSize),
bodyStream: file.openRead(),
progressTotal: totalSize,
onProgress: (sent, total) {
if (!cancelled) ctrl.add(UploadProgress(sent: sent, total: total));
},
progressThrottle: progressThrottle,
autoForceAfter: autoForceAfter,
timeout: overallTimeout,
onSocketReady: (s) => socket = s,
shouldAbort: () => cancelled,
);
var status = 0;
String? error;
final done = Completer<void>();
sub =
session
.uploadFilePath(
url: info.url,
path: file.path,
filename: filename,
)
.listen(
(e) {
switch (e) {
case kb.UploadEvent_Progress(:final sent, :final total):
ctrl.add(
UploadProgress(
sent: sent.toInt(),
total: total.toInt(),
),
);
case kb.UploadEvent_Done(status: final s):
status = s;
case kb.UploadEvent_Error(:final message):
error = message;
}
},
onError: (Object err) {
error = err.toString();
if (!done.isCompleted) done.complete();
},
onDone: () {
if (!done.isCompleted) done.complete();
},
cancelOnError: true,
);
await done.future;
if (cancelled) return;
final statusCode = result?.$1 ?? 0;
if (statusCode != 200 && statusCode != 0) {
ctrl.add(UploadError('http_$statusCode'));
if (error != null) {
ctrl.add(UploadError(error!));
return;
}
if (status != 200 && status != 0) {
ctrl.add(UploadError('http_$status'));
return;
}
@@ -137,9 +162,6 @@ class FileUploader {
} catch (e) {
if (!cancelled) ctrl.add(UploadError(e.toString()));
} finally {
try {
socket?.destroy();
} catch (_) {}
await ctrl.close();
}
}
@@ -155,133 +177,51 @@ class FileUploader {
Duration overallTimeout = const Duration(minutes: 5),
Duration progressThrottle = const Duration(milliseconds: 16),
}) async {
final session = api.session;
if (session == null) return false;
try {
final total = await file.length();
if (total <= 0) return false;
final filename = _syntheticFilename();
final result = await _sendHttpRequest(
uri,
method: 'POST',
headers: _buildUploadHeaders(
uri,
filename,
total,
contentType: 'application/octet-stream',
connection: 'close',
final result = await _consume(
session.uploadFilePath(
url: uri.toString(),
path: file.path,
filename: _syntheticFilename(),
),
bodyStream: file.openRead(),
progressTotal: total,
onProgress: onProgress,
progressThrottle: progressThrottle,
timeout: overallTimeout,
);
final statusCode = result?.$1 ?? 0;
final respBody = result?.$2 ?? '';
logger.w(
'uploadMediaFile: status=$statusCode total=$total '
'host=${uri.host} body=${respBody.length > 200 ? respBody.substring(0, 200) : respBody}',
);
if (result.error != null) {
logger.w('uploadMediaFile: ${result.error}');
return false;
}
final respBody = utf8.decode(result.body, allowMalformed: true);
final hasError =
respBody.contains('error_msg') || respBody.contains('error_code');
return statusCode == 200 && !hasError;
return result.status == 200 && !hasError;
} catch (e) {
logger.w('uploadMediaFile: $e');
return false;
}
}
Future<Socket> _openSocket(Uri uri) async {
final proxySettings = await ProxyConfig.load();
final base = proxySettings.isEnabled
? await ProxyConnector(proxySettings).connect(uri.host, uri.port)
: await Socket.connect(uri.host, uri.port);
if (uri.scheme != 'https') return base;
final allowInsecure = await TlsConfig.isInsecureAllowed();
if (allowInsecure) {
logger.w(
'TLS: проверка сертификата отключена (дебаг) — загрузка уязвима к MitM',
);
return SecureSocket.secure(
base,
host: uri.host,
onBadCertificate: (_) => true,
);
}
return SecureSocket.secure(base, host: uri.host);
}
String _syntheticFilename() =>
(DateTime.now().microsecondsSinceEpoch & 0x7FFFFFFF).toString();
String _multipartBoundary() =>
'----KometBoundary${DateTime.now().microsecondsSinceEpoch}';
Map<String, String> _buildUploadHeaders(
Uri uri,
String filename,
int total, {
String contentType = 'application/x-binary; charset=x-user-defined',
String connection = 'keep-alive',
}) {
return {
'Host': uri.host,
'Content-Type': contentType,
'Content-Disposition': 'attachment; filename=$filename',
'Connection': connection,
'User-Agent': Uri.encodeComponent(_userAgentHeader),
'Content-Range': 'bytes 0-${total - 1}/$total',
'Content-Length': '$total',
};
}
Future<String?> uploadImage(
Uri uri,
Uint8List bytes, {
String filename = 'avatar.jpg',
}) async {
final session = api.session;
if (session == null) return null;
try {
final boundary = _multipartBoundary();
final preamble = utf8.encode(
'--$boundary\r\n'
'Content-Disposition: form-data; name="file"; filename="$filename"\r\n'
'Content-Type: ${_contentTypeForFilename(filename)}\r\n'
'\r\n',
);
final epilogue = utf8.encode('\r\n--$boundary--\r\n');
final response = await _sendHttpRequest(
uri,
method: 'POST',
headers: _buildMultipartHeaders(
uri,
preamble.length + bytes.length + epilogue.length,
boundary: boundary,
final result = await _consume(
session.uploadPhoto(
url: uri.toString(),
data: bytes,
filename: filename,
),
prefixBytes: preamble,
bodyStream: Stream.value(bytes),
suffixBytes: epilogue,
timeout: const Duration(minutes: 2),
);
if (response == null) {
if (result.error != null || result.status != 200) {
logger.w('uploadImage: status=${result.status} error=${result.error}');
return null;
}
final (status, body) = response;
if (status != 200) {
logger.w(
'uploadImage: status=$status body=${body.length > 200 ? '${body.substring(0, 200)}…' : body}',
);
return null;
}
final token = _parsePhotoToken(body);
if (token == null) {
logger.w(
'uploadImage: photoToken not found in body=${body.length > 200 ? '${body.substring(0, 200)}…' : body}',
);
}
return token;
return _parsePhotoToken(utf8.decode(result.body, allowMalformed: true));
} catch (e) {
logger.w('uploadImage: $e');
return null;
@@ -295,41 +235,22 @@ class FileUploader {
void Function(int sent, int total)? onProgress,
Duration progressThrottle = const Duration(milliseconds: 16),
}) async {
final session = api.session;
if (session == null) return null;
try {
final fileLength = await file.length();
final boundary = _multipartBoundary();
final preamble = utf8.encode(
'--$boundary\r\n'
'Content-Disposition: form-data; name="file"; filename="$filename"\r\n'
'Content-Type: ${_contentTypeForFilename(filename)}\r\n'
'\r\n',
);
final epilogue = utf8.encode('\r\n--$boundary--\r\n');
final response = await _sendHttpRequest(
uri,
method: 'POST',
headers: _buildMultipartHeaders(
uri,
preamble.length + fileLength + epilogue.length,
boundary: boundary,
final result = await _consume(
session.uploadPhotoPath(
url: uri.toString(),
path: file.path,
filename: filename,
),
prefixBytes: preamble,
bodyStream: file.openRead(),
suffixBytes: epilogue,
progressTotal: fileLength,
onProgress: onProgress,
progressThrottle: progressThrottle,
timeout: const Duration(minutes: 2),
);
if (response == null) return null;
final (status, responseBody) = response;
if (status != 200) {
logger.w('uploadPhoto: status=$status');
if (result.error != null || result.status != 200) {
logger.w('uploadPhoto: status=${result.status} error=${result.error}');
return null;
}
return _parsePhotoToken(responseBody);
return _parsePhotoToken(utf8.decode(result.body, allowMalformed: true));
} catch (e) {
logger.w('uploadPhoto: $e');
return null;
@@ -344,342 +265,50 @@ class FileUploader {
int concurrency = 4,
Duration overallTimeout = const Duration(minutes: 30),
}) async {
final total = await file.length();
if (total <= 0) return false;
final fileName = _syntheticFilename();
final handshake = await _okCdnRequest(
uri,
method: 'GET',
fileName: fileName,
timeout: const Duration(seconds: 30),
);
if (handshake == null || handshake.$1 != 200) return false;
var startOffset = 0;
final resumed = int.tryParse(handshake.$2.trim());
if (resumed != null && resumed > 0 && resumed <= total) {
startOffset = resumed;
}
final ranges = <(int, int)>[];
for (var o = startOffset; o < total; o += chunkSize) {
ranges.add((o, o + chunkSize < total ? o + chunkSize : total));
}
if (ranges.isEmpty) return true;
var nextIndex = 0;
var sent = startOffset;
var failed = false;
Future<void> worker() async {
while (!failed) {
final i = nextIndex++;
if (i >= ranges.length) return;
final (start, end) = ranges[i];
final bytes = await _readRange(file, start, end);
final resp = await _okCdnRequest(
uri,
method: 'POST',
fileName: fileName,
body: bytes,
contentRange: 'bytes $start-${end - 1}/$total',
timeout: overallTimeout,
);
if (resp == null || (resp.$1 != 200 && resp.$1 != 201)) {
logger.w('uploadVideoFile: chunk status=${resp?.$1}');
failed = true;
return;
}
sent += end - start;
onProgress?.call(sent, total);
}
}
final workerCount = concurrency < ranges.length
? concurrency
: ranges.length;
await Future.wait(List.generate(workerCount, (_) => worker()));
return !failed;
}
Future<Uint8List> _readRange(File file, int start, int end) async {
final builder = BytesBuilder(copy: false);
await for (final chunk in file.openRead(start, end)) {
builder.add(chunk);
}
return builder.takeBytes();
}
Future<(int, String)?> _okCdnRequest(
Uri uri, {
required String method,
required String fileName,
Uint8List? body,
String? contentRange,
required Duration timeout,
}) async {
final session = api.session;
if (session == null) return false;
try {
final headers = {
'Host': uri.host,
'Content-Type': 'application/x-binary; charset=x-user-defined',
'Content-Disposition': 'attachment; fileName="$fileName"',
'Content-Range': ?contentRange,
'Content-Length': '${body?.length ?? 0}',
'X-Uploading-Mode': 'parallel',
'Connection': 'close',
};
return await _sendHttpRequest(
uri,
method: method,
headers: headers,
prefixBytes: body,
timeout: timeout,
final result = await _consume(
session.uploadVideoPath(
url: uri.toString(),
path: file.path,
chunkSize: chunkSize,
concurrency: concurrency,
),
onProgress: onProgress,
);
return result.error == null && result.status == 200;
} catch (e) {
logger.w('_okCdnRequest($method): $e');
return null;
logger.w('uploadVideoFile: $e');
return false;
}
}
Map<String, String> _buildMultipartHeaders(
Uri uri,
int total, {
required String boundary,
}) {
return {
'Host': uri.host,
'Content-Type': 'multipart/form-data; boundary=$boundary',
'Content-Length': '$total',
'Connection': 'keep-alive',
'User-Agent': Uri.encodeComponent(_userAgentHeader),
};
}
Future<(int, String)?> _sendHttpRequest(
Uri uri, {
required String method,
required Map<String, String> headers,
List<int>? prefixBytes,
Stream<List<int>>? bodyStream,
List<int>? suffixBytes,
int? progressTotal,
/// Прогоняет стрим ядра до конца, форвардит прогресс, отдаёт итог.
Future<({int status, Uint8List body, String? error})> _consume(
Stream<kb.UploadEvent> stream, {
void Function(int sent, int total)? onProgress,
Duration progressThrottle = const Duration(milliseconds: 16),
Duration? autoForceAfter,
required Duration timeout,
void Function(Socket socket)? onSocketReady,
bool Function()? shouldAbort,
}) async {
final socket = await _openSocket(uri);
onSocketReady?.call(socket);
try {
if (shouldAbort?.call() ?? false) return null;
_writeRequestHeaders(socket, uri, method, headers);
if (prefixBytes != null && prefixBytes.isNotEmpty) {
socket.add(prefixBytes);
}
if (bodyStream != null) {
final stream = (onProgress != null && progressTotal != null)
? _withProgress(
bodyStream,
progressTotal,
onProgress,
throttle: progressThrottle,
)
: bodyStream;
await socket.addStream(stream);
}
if (suffixBytes != null && suffixBytes.isNotEmpty) {
socket.add(suffixBytes);
}
await socket.flush();
if (onProgress != null && progressTotal != null) {
onProgress(progressTotal, progressTotal);
}
if (shouldAbort?.call() ?? false) return null;
if (autoForceAfter != null) {
final status = await _readResponse(
socket,
autoForceAfter: autoForceAfter,
overallTimeout: timeout,
);
return (status, '');
}
return await _readFullResponse(socket, timeout: timeout);
} finally {
try {
socket.destroy();
} catch (_) {}
}
}
void _writeRequestHeaders(
Socket socket,
Uri uri,
String method,
Map<String, String> headers,
) {
final path = '${uri.path}${uri.hasQuery ? "?${uri.query}" : ""}';
final buffer = StringBuffer()..write('$method $path HTTP/1.1\r\n');
for (final entry in headers.entries) {
buffer.write('${entry.key}: ${entry.value}\r\n');
}
buffer.write('\r\n');
socket.add(utf8.encode(buffer.toString()));
}
Stream<List<int>> _withProgress(
Stream<List<int>> src,
int total,
void Function(int sent, int total) onProgress, {
Duration throttle = const Duration(milliseconds: 16),
}) {
final stopwatch = Stopwatch()..start();
var sent = 0;
return src.map((chunk) {
sent += chunk.length;
if (stopwatch.elapsed >= throttle) {
onProgress(sent, total);
stopwatch.reset();
}
return chunk;
});
}
String _contentTypeForFilename(String filename) {
final ext = filename.contains('.')
? filename.split('.').last.toLowerCase()
: '';
switch (ext) {
case 'png':
return 'image/png';
case 'gif':
return 'image/gif';
case 'webp':
return 'image/webp';
case 'heic':
case 'heif':
return 'image/heic';
case 'bmp':
return 'image/bmp';
case 'jpg':
case 'jpeg':
default:
return 'image/jpeg';
}
}
Future<(int, String)?> _readFullResponse(
Socket socket, {
required Duration timeout,
}) {
final bytes = <int>[];
final completer = Completer<(int, String)?>();
Timer? timer;
StreamSubscription<List<int>>? sub;
void finishWith((int, String)? value) {
timer?.cancel();
sub?.cancel();
if (!completer.isCompleted) completer.complete(value);
}
(int, String)? tryParse({required bool atClose}) {
final headerEnd = _findHeaderEnd(bytes);
if (headerEnd == -1) return null;
final headerStr = utf8.decode(
bytes.sublist(0, headerEnd),
allowMalformed: true,
);
final lines = headerStr.split('\r\n');
final parts = lines.first.split(' ');
final status = parts.length >= 2 ? (int.tryParse(parts[1]) ?? 0) : 0;
final headerLines = lines.skip(1);
final chunked = headerLines.any(
(l) =>
l.toLowerCase().startsWith('transfer-encoding:') &&
l.toLowerCase().contains('chunked'),
);
int? contentLength;
for (final l in headerLines) {
if (l.toLowerCase().startsWith('content-length:')) {
contentLength = int.tryParse(l.split(':').last.trim());
}
}
final rawBody = utf8.decode(
bytes.sublist(headerEnd),
allowMalformed: true,
);
if (chunked) {
if (!atClose && !rawBody.contains('\r\n0\r\n')) return null;
return (status, _decodeChunked(rawBody));
}
if (contentLength != null &&
!atClose &&
bytes.length - headerEnd < contentLength) {
return null;
}
return (status, rawBody);
}
sub = socket.listen(
(chunk) {
bytes.addAll(chunk);
final parsed = tryParse(atClose: false);
if (parsed != null) finishWith(parsed);
},
onError: (e) {
logger.w('uploadImage: socket error after ${bytes.length} bytes: $e');
finishWith(tryParse(atClose: true));
},
onDone: () {
final parsed = tryParse(atClose: true);
if (parsed == null) {
logger.w(
'uploadImage: connection closed without HTTP response (${bytes.length} bytes)',
);
}
finishWith(parsed);
},
);
timer = Timer(timeout, () {
logger.w('uploadImage: response timeout after ${bytes.length} bytes');
finishWith(tryParse(atClose: true));
});
return completer.future;
}
String _decodeChunked(String body) {
final out = StringBuffer();
var i = 0;
while (i < body.length) {
final lineEnd = body.indexOf('\r\n', i);
if (lineEnd < 0) break;
final sizeStr = body.substring(i, lineEnd).split(';').first.trim();
if (sizeStr.isEmpty) {
i = lineEnd + 2;
continue;
}
final size = int.tryParse(sizeStr, radix: 16);
if (size == null) break;
if (size == 0) break;
final dataStart = lineEnd + 2;
if (dataStart + size > body.length) break;
out.write(body.substring(dataStart, dataStart + size));
i = dataStart + size;
if (i + 2 <= body.length && body.substring(i, i + 2) == '\r\n') {
i += 2;
var status = 0;
var body = Uint8List(0);
String? error;
await for (final event in stream) {
switch (event) {
case kb.UploadEvent_Progress(:final sent, :final total):
onProgress?.call(sent.toInt(), total.toInt());
case kb.UploadEvent_Done(status: final s, body: final b):
status = s;
body = b;
case kb.UploadEvent_Error(:final message):
error = message;
}
}
return out.toString();
return (status: status, body: body, error: error);
}
String _syntheticFilename() =>
(DateTime.now().microsecondsSinceEpoch & 0x7FFFFFFF).toString();
String? _parsePhotoToken(String body) {
try {
final json = jsonDecode(body);
@@ -701,79 +330,4 @@ class FileUploader {
}
return null;
}
Future<int> _readResponse(
Socket socket, {
required Duration autoForceAfter,
required Duration overallTimeout,
}) {
final responseBytes = <int>[];
final completer = Completer<int>();
Timer? force;
Timer? overall;
StreamSubscription<List<int>>? sub;
void finish(int code) {
if (completer.isCompleted) return;
force?.cancel();
overall?.cancel();
sub?.cancel();
completer.complete(code);
}
void fail(Object e) {
if (completer.isCompleted) return;
force?.cancel();
overall?.cancel();
sub?.cancel();
completer.completeError(e);
}
force = Timer(autoForceAfter, () => finish(0));
sub = socket.listen(
responseBytes.addAll,
onError: fail,
onDone: () {
final code = _parseHttpStatus(responseBytes);
if (code == null) {
fail(const SocketException('Не удалось прочитать заголовок ответа'));
} else {
finish(code);
}
},
);
overall = Timer(
overallTimeout,
() => fail(TimeoutException('Тайм-аут загрузки')),
);
return completer.future;
}
int? _parseHttpStatus(List<int> bytes) {
final headerEnd = _findHeaderEnd(bytes);
if (headerEnd == -1) return null;
final headerStr = utf8.decode(
bytes.sublist(0, headerEnd),
allowMalformed: true,
);
final statusLine = headerStr.split('\r\n').first;
final parts = statusLine.split(' ');
if (parts.length < 2) return null;
return int.tryParse(parts[1]);
}
int _findHeaderEnd(List<int> bytes) {
for (var i = 0; i < bytes.length - 3; i++) {
if (bytes[i] == 0x0D &&
bytes[i + 1] == 0x0A &&
bytes[i + 2] == 0x0D &&
bytes[i + 3] == 0x0A) {
return i + 4;
}
}
return -1;
}
}
lib/core/protocol/packet.dart
+4 -152
View File
@@ -1,16 +1,3 @@
import 'dart:typed_data';
import 'dart:isolate';
import 'package:dart_lz4/dart_lz4.dart';
import 'package:libcompress/libcompress.dart';
import 'package:msgpack_dart/msgpack_dart.dart' as msgpack;
import 'lz4_block.dart';
/// ver(1) + cmd(1) + seq(2) + opcode(2) + packedLen(4) = 10
const int headerSize = 10;
/// Потолок распаковки payload (анти-бомба); буфер растёт динамически до него.
const int _maxDecompressedSize = 32 * 1024 * 1024; // 32 MB
/// Типы команд в протоколе
abstract class CmdType {
static const int request =
@@ -22,17 +9,11 @@ abstract class CmdType {
static const int error = 3; // ответ: ошибка
}
/// Распакованный бинарный пакет
/// Распакованный пакет.
///
/// Формат заголовка (10 байт):
/// ```
/// [0] ver — версия протокола (uint8) (по умолчанию 10)
/// [1] cmd — тип команды (uint8) (при отправке от клиента равно 0)
/// [2..3] seq — порядковый номер (uint16 BE)
/// [4..5] opcode — код операции (uint16 BE)
/// [6..9] packedLen — флаг сжатия [6] + длина payload [7..9] (uint32 BE)
/// [10..] payload — данные в MsgPack, опционально сжатые LZ4
/// ```
/// Провод (фрейминг, MsgPack, сжатие) живёт в Rust-ядре kolibri; здесь пакет —
/// это уже декодированный [payload] (Map/List/скаляр, бинарь — Uint8List) плюс
/// метаданные заголовка.
class Packet {
int api;
int cmd;
@@ -109,132 +90,3 @@ bool isSessionStateError(Object error) {
text.contains('авторизационная сессия') ||
text.contains('сессия не онлайн');
}
/// Payload меньше этого размера отправляется без сжатия (как в оригинале).
const int _compressionThreshold = 32;
/// Упаковка пакета для отправки на сервер.
///
/// Payload сериализуется в MsgPack и при размере >= [_compressionThreshold]
/// сжимается LZ4-block. Старший байт поля packedLen — флаг сжатия:
/// `0` — без сжатия, иначе `(rawLen ~/ compLen) + 1` (множитель размера, по
/// которому получатель выделяет буфер под распаковку).
Uint8List packPacket(int opcode, Map<dynamic, dynamic> payload, {int seq = 0}) {
final Uint8List raw = msgpack.serialize(payload);
final List<int> body;
final int flag;
if (raw.length < _compressionThreshold) {
body = raw;
flag = 0;
} else {
body = lz4Compress(raw);
flag = (raw.length ~/ body.length) + 1;
}
final out = Uint8List(headerSize + body.length);
final header = ByteData.view(out.buffer, out.offsetInBytes, headerSize);
header.setUint8(0, 10);
header.setUint8(1, CmdType.request);
header.setUint16(2, seq, Endian.big);
header.setUint16(4, opcode, Endian.big);
header.setUint32(
6,
((flag & 0xFF) << 24) | (body.length & 0xFFFFFF),
Endian.big,
);
out.setRange(headerSize, out.length, body);
return out;
}
const int _isolateDecodeThreshold = 4096;
Future<Packet> unpackPacket(Uint8List packet) async {
final header = ByteData.sublistView(packet);
final apiVer = header.getUint8(0) & 0xFF;
final cmd = header.getUint8(1) & 0xFF;
final seq = header.getUint16(2) & 0xFFFF;
final opcode = header.getUint16(4) & 0xFFFF;
final packedLen = header.getUint32(6);
final compFlag = packedLen >> 24;
final payloadLength = packedLen & 0xFFFFFF;
if (payloadLength == 0) {
return Packet(api: apiVer, cmd: cmd, seq: seq, opcode: opcode);
}
final end = headerSize + payloadLength;
if (end > packet.length) {
throw Exception('Packet payload length $payloadLength exceeds buffer');
}
final slice = Uint8List.sublistView(packet, headerSize, end);
dynamic payload;
if (compFlag == 0 && slice.length < _isolateDecodeThreshold) {
payload = _deserializePayload(slice, compFlag);
} else {
final owned = Uint8List.fromList(slice);
payload = await Isolate.run(() => _deserializePayload(owned, compFlag));
}
return Packet(
api: apiVer,
cmd: cmd,
seq: seq,
opcode: opcode,
payload: payload,
);
}
dynamic _deserializePayload(Uint8List payloadBytes, int compFlag) {
var bytes = payloadBytes;
if (compFlag != 0) {
bytes = _decompressPayload(bytes);
}
if (bytes.isEmpty) return null;
try {
return msgpack.deserialize(bytes);
} catch (e) {
throw Exception('MsgPack deserialization error: $e');
}
}
/// Определяет формат сжатия по 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 {
return ZstdCodec(
maxDecompressedSize: _maxDecompressedSize,
).decompress(src);
} 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');
}
}
lib/main.dart
+2
View File
@@ -4,6 +4,7 @@ import 'dart:ui' as ui;
import 'package:dynamic_color/dynamic_color.dart';
import 'package:flutter/cupertino.dart' show CupertinoPageTransitionsBuilder;
import 'package:kolibri/kolibri.dart' show initKolibri;
import 'package:flutter/material.dart';
import 'package:flutter/rendering.dart';
import 'package:video_player_media_kit/video_player_media_kit.dart';
@@ -156,6 +157,7 @@ void _installLogCapture() {
void main(List<String> args) async {
WidgetsFlutterBinding.ensureInitialized();
await initKolibri();
DebugTest.parse(args);
_installLogCapture();
VideoPlayerMediaKit.ensureInitialized(
macos/Podfile.lock
+12 -1
View File
@@ -84,6 +84,8 @@ PODS:
- GoogleUtilities/UserDefaults (8.1.1):
- GoogleUtilities/Logger
- GoogleUtilities/Privacy
- kolibri (0.0.1):
- FlutterMacOS
- media_kit_libs_macos_video (1.0.4):
- FlutterMacOS
- media_kit_video (0.0.1):
@@ -108,6 +110,7 @@ PODS:
- PromisesObjC (2.4.1)
- record_macos (1.2.1):
- FlutterMacOS
- rlottie (0.2.0)
- share_plus (0.0.1):
- FlutterMacOS
- shared_preferences_foundation (0.0.1):
@@ -139,6 +142,7 @@ DEPENDENCIES:
- flutter_webrtc (from `Flutter/ephemeral/.symlinks/plugins/flutter_webrtc/macos`)
- FlutterMacOS (from `Flutter/ephemeral`)
- geolocator_apple (from `Flutter/ephemeral/.symlinks/plugins/geolocator_apple/darwin`)
- kolibri (from `Flutter/ephemeral/.symlinks/plugins/kolibri/macos`)
- media_kit_libs_macos_video (from `Flutter/ephemeral/.symlinks/plugins/media_kit_libs_macos_video/macos`)
- media_kit_video (from `Flutter/ephemeral/.symlinks/plugins/media_kit_video/macos`)
- mobile_scanner (from `Flutter/ephemeral/.symlinks/plugins/mobile_scanner/darwin`)
@@ -146,6 +150,7 @@ DEPENDENCIES:
- package_info_plus (from `Flutter/ephemeral/.symlinks/plugins/package_info_plus/macos`)
- photo_manager (from `Flutter/ephemeral/.symlinks/plugins/photo_manager/darwin`)
- record_macos (from `Flutter/ephemeral/.symlinks/plugins/record_macos/macos`)
- rlottie (from `../third_party`)
- share_plus (from `Flutter/ephemeral/.symlinks/plugins/share_plus/macos`)
- shared_preferences_foundation (from `Flutter/ephemeral/.symlinks/plugins/shared_preferences_foundation/darwin`)
- sqflite_darwin (from `Flutter/ephemeral/.symlinks/plugins/sqflite_darwin/darwin`)
@@ -194,6 +199,8 @@ EXTERNAL SOURCES:
:path: Flutter/ephemeral
geolocator_apple:
:path: Flutter/ephemeral/.symlinks/plugins/geolocator_apple/darwin
kolibri:
:path: Flutter/ephemeral/.symlinks/plugins/kolibri/macos
media_kit_libs_macos_video:
:path: Flutter/ephemeral/.symlinks/plugins/media_kit_libs_macos_video/macos
media_kit_video:
@@ -208,6 +215,8 @@ EXTERNAL SOURCES:
:path: Flutter/ephemeral/.symlinks/plugins/photo_manager/darwin
record_macos:
:path: Flutter/ephemeral/.symlinks/plugins/record_macos/macos
rlottie:
:path: "../third_party"
share_plus:
:path: Flutter/ephemeral/.symlinks/plugins/share_plus/macos
shared_preferences_foundation:
@@ -242,6 +251,7 @@ SPEC CHECKSUMS:
geolocator_apple: ab36aa0e8b7d7a2d7639b3b4e48308394e8cef5e
GoogleDataTransport: aae35b7ea0c09004c3797d53c8c41f66f219d6a7
GoogleUtilities: 4f2618a4a1e762a1ee134a1e2323bba9843e06da
kolibri: 93062ece67f68ec0b909876b527aa15e198b4a73
media_kit_libs_macos_video: 85a23e549b5f480e72cae3e5634b5514bc692f65
media_kit_video: fa6564e3799a0a28bff39442334817088b7ca758
mobile_scanner: 9157936403f5a0644ca3779a38ff8404c5434a93
@@ -252,6 +262,7 @@ SPEC CHECKSUMS:
photo_manager: 25fd77df14f4f0ba5ef99e2c61814dde77e2bceb
PromisesObjC: 752c3227f599e3467650e47ea36f433eeb10c273
record_macos: 5d55909f9650314be6424ffd6b123ac75a08c3c1
rlottie: c2780aabf858ca3c81785f09256827ca83b0765d
share_plus: 510bf0af1a42cd602274b4629920c9649c52f4cc
shared_preferences_foundation: 7036424c3d8ec98dfe75ff1667cb0cd531ec82bb
sqflite_darwin: 20b2a3a3b70e43edae938624ce550a3cbf66a3d0
@@ -260,6 +271,6 @@ SPEC CHECKSUMS:
wakelock_plus: 917609be14d812ddd9e9528876538b2263aaa03b
WebRTC-SDK: e6006119cd730d6315d875e4a421b6cc8bb88833
PODFILE CHECKSUM: 54d867c82ac51cbd61b565781b9fada492027009
PODFILE CHECKSUM: f9f028435b3eb11958579119e016406574b87759
COCOAPODS: 1.16.2
pubspec.lock
+31 -16
View File
@@ -81,6 +81,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.1.2"
build_cli_annotations:
dependency: transitive
description:
name: build_cli_annotations
sha256: e563c2e01de8974566a1998410d3f6f03521788160a02503b0b1f1a46c7b3d95
url: "https://pub.dev"
source: hosted
version: "2.1.1"
button_group_m3e:
dependency: transitive
description:
@@ -499,6 +507,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.0.35"
flutter_rust_bridge:
dependency: transitive
description:
name: flutter_rust_bridge
sha256: e87d6b9ee934dcd24a128ccb2bd91905d2d5fe5c06245d6a8f5477d4907a437a
url: "https://pub.dev"
source: hosted
version: "2.12.0"
flutter_secure_storage:
dependency: "direct main"
description:
@@ -581,6 +597,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "1.5.2"
freezed_annotation:
dependency: transitive
description:
name: freezed_annotation
sha256: c2e2d632dd9b8a2b7751117abcfc2b4888ecfe181bd9fca7170d9ef02e595fe2
url: "https://pub.dev"
source: hosted
version: "2.4.4"
geolocator:
dependency: "direct main"
description:
@@ -733,6 +757,13 @@ packages:
url: "https://pub.dev"
source: hosted
version: "4.12.0"
kolibri:
dependency: "direct main"
description:
path: "third_party/kolibri/kolibri-dart"
relative: true
source: path
version: "0.1.0"
leak_tracker:
dependency: transitive
description:
@@ -757,14 +788,6 @@ packages:
url: "https://pub.dev"
source: hosted
version: "3.0.2"
libcompress:
dependency: "direct main"
description:
name: libcompress
sha256: "1f55be8dc9e622efa1584ad899e05880d71469b7107f35be1858e91d0fadf1d4"
url: "https://pub.dev"
source: hosted
version: "1.0.0"
lints:
dependency: transitive
description:
@@ -917,14 +940,6 @@ packages:
url: "https://pub.dev"
source: hosted
version: "7.2.0"
msgpack_dart:
dependency: "direct main"
description:
name: msgpack_dart
sha256: c2d235ed01f364719b5296aecf43ac330f0d7bc865fa134d0d7910a40454dffb
url: "https://pub.dev"
source: hosted
version: "1.0.1"
native_toolchain_c:
dependency: transitive
description:
pubspec.yaml
+5 -2
View File
@@ -34,11 +34,14 @@ dependencies:
sdk: flutter
intl: any
# Rust networking core (kolibri) — FFI plugin, replaces the Dart transport.
# Vendored as a git submodule (third_party/kolibri).
kolibri:
path: third_party/kolibri/kolibri-dart
# The following adds the Cupertino Icons font to your application.
# Use with the CupertinoIcons class for iOS style icons.
dart_lz4: ^1.0.0
libcompress: ^1.0.0
msgpack_dart: ^1.0.1
crypto: ^3.0.7
ffi: ^2.1.0
logger: ^2.6.2