HyperNet/Messaging
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Basic websocket connection

Browse Source
This commit is contained in:
LittleSheep
2024-11-15 00:24:46 +08:00
parent 2e68d227a0
commit 8bc0da5188
11 changed files with 472 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
lib/providers/sn_network.dart
View File

@ -44,48 +44,11 @@ class SnNetworkProvider {
RequestOptions options,
RequestInterceptorHandler handler,
) async {
try {
var atk = await _storage.read(key: kAtkStoreKey);
if (atk != null) {
final atkParts = atk.split('.');
if (atkParts.length != 3) {
throw Exception('invalid format of access token');
}
var rawPayload =
atkParts[1].replaceAll('-', '+').replaceAll('_', '/');
switch (rawPayload.length % 4) {
case 0:
break;
case 2:
rawPayload += '==';
break;
case 3:
rawPayload += '=';
break;
default:
throw Exception('illegal format of access token payload');
}
final b64 = utf8.fuse(base64Url);
final payload = b64.decode(rawPayload);
final exp = jsonDecode(payload)['exp'];
if (exp <= DateTime.now().millisecondsSinceEpoch ~/ 1000) {
log('Access token need refresh, doing it at ${DateTime.now()}');
atk = await refreshToken();
}
if (atk != null) {
options.headers['Authorization'] = 'Bearer $atk';
} else {
log('Access token refresh failed...');
}
}
} catch (err) {
log('Failed to authenticate user: $err');
} finally {
handler.next(options);
final atk = await getFreshAtk();
if (atk != null) {
options.headers['Authorization'] = 'Bearer $atk';
}
return handler.next(options);
},
),
);
@ -99,6 +62,50 @@ class SnNetworkProvider {
});
}
Future<String?> getFreshAtk() async {
try {
var atk = await _storage.read(key: kAtkStoreKey);
if (atk != null) {
final atkParts = atk.split('.');
if (atkParts.length != 3) {
throw Exception('invalid format of access token');
}
var rawPayload = atkParts[1].replaceAll('-', '+').replaceAll('_', '/');
switch (rawPayload.length % 4) {
case 0:
break;
case 2:
rawPayload += '==';
break;
case 3:
rawPayload += '=';
break;
default:
throw Exception('illegal format of access token payload');
}
final b64 = utf8.fuse(base64Url);
final payload = b64.decode(rawPayload);
final exp = jsonDecode(payload)['exp'];
if (exp <= DateTime.now().millisecondsSinceEpoch ~/ 1000) {
log('Access token need refresh, doing it at ${DateTime.now()}');
atk = await refreshToken();
}
if (atk != null) {
return atk;
} else {
log('Access token refresh failed...');
}
}
} catch (err) {
log('Failed to authenticate user: $err');
}
return null;
}
String getAttachmentUrl(String ky) {
if (ky.startsWith("http")) return ky;
return '${client.options.baseUrl}/cgi/uc/attachments/$ky';

2
lib/providers/userinfo.dart
View File

@ -13,6 +13,8 @@ class UserProvider extends ChangeNotifier {
late final SnNetworkProvider _sn;
late final FlutterSecureStorage _storage = FlutterSecureStorage();
Future<String?> get atk => _storage.read(key: kAtkStoreKey);
UserProvider(BuildContext context) {
_sn = context.read<SnNetworkProvider>();

109
lib/providers/websocket.dart Normal file
View File

@ -0,0 +1,109 @@
import 'dart:async';
import 'dart:convert';
import 'dart:developer';
import 'package:flutter/material.dart';
import 'package:provider/provider.dart';
import 'package:surface/providers/sn_network.dart';
import 'package:surface/providers/userinfo.dart';
import 'package:surface/types/websocket.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
class WebSocketProvider extends ChangeNotifier {
bool isBusy = false;
bool isConnected = false;
WebSocketChannel? conn;
late final SnNetworkProvider _sn;
late final UserProvider _ua;
StreamController<WebSocketPackage> stream = StreamController.broadcast();
WebSocketProvider(BuildContext context) {
_sn = context.read<SnNetworkProvider>();
_ua = context.read<UserProvider>();
// Wait for the userinfo provide initialize authorization status
Future.delayed(const Duration(milliseconds: 250), () async {
if (_ua.isAuthorized) {
log('[WebSocket] Connecting to the server...');
await connect();
} else {
log('[WebSocket] Unable connect to the server, unauthorized.');
}
});
}
Future<void> connect({noRetry = false}) async {
if (!_ua.isAuthorized) return;
if (isConnected) {
disconnect();
}
final atk = await _sn.getFreshAtk();
final uri = Uri.parse(
'${_sn.client.options.baseUrl.replaceFirst('http', 'ws')}/ws?tk=$atk',
);
isBusy = true;
notifyListeners();
try {
conn = WebSocketChannel.connect(uri);
await conn!.ready;
log('[WebSocket] Connected to server!');
} catch (err) {
if (err is WebSocketChannelException) {
log('Failed to connect to websocket: ${(err.inner as dynamic).message}');
} else {
log('Failed to connect to websocket: $err');
}
if (!noRetry) {
log('Retry connecting to websocket in 3 seconds...');
return Future.delayed(
const Duration(seconds: 3),
() => connect(noRetry: true),
);
}
} finally {
isBusy = false;
notifyListeners();
}
}
void disconnect() {
if (conn != null) {
conn!.sink.close();
}
isConnected = false;
notifyListeners();
}
void listen() {
conn?.stream.listen(
(event) {
final packet = WebSocketPackage.fromJson(jsonDecode(event));
log('Websocket incoming message: ${packet.method} ${packet.message}');
stream.sink.add(packet);
// TODO handle notification
// if (packet.method == 'notifications.new') {
// final NotificationProvider nty = Get.find();
// nty.notifications.add(Notification.fromJson(packet.payload!));
// nty.notificationUnread.value++;
// }
},
onDone: () {
isConnected = false;
notifyListeners();
Future.delayed(const Duration(seconds: 1), () => connect());
},
onError: (err) {
isConnected = false;
notifyListeners();
Future.delayed(const Duration(seconds: 11), () => connect());
},
);
}
}