실시간 구독 공유 스트림 — 여러 화면이 원본 구독 하나를 나눠 쓴다 · 서버 시험을 저장소 안 에뮬레이터 도구로 실행
This commit is contained in:
@@ -12,7 +12,7 @@
|
|||||||
"build": "tsc",
|
"build": "tsc",
|
||||||
"typecheck": "tsc --noEmit",
|
"typecheck": "tsc --noEmit",
|
||||||
"test": "vitest run",
|
"test": "vitest run",
|
||||||
"test:emulator": "firebase emulators:exec --config ../firebase.json --only firestore --project demo-fanuzu \"npm test\""
|
"test:emulator": "../test/node_modules/.bin/firebase emulators:exec --config ../firebase.json --only firestore --project demo-fanuzu \"npm test\""
|
||||||
},
|
},
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"firebase-admin": "13.6.0",
|
"firebase-admin": "13.6.0",
|
||||||
|
|||||||
@@ -0,0 +1,75 @@
|
|||||||
|
import 'dart:async';
|
||||||
|
|
||||||
|
/// 원본 구독 하나를 여러 화면이 나눠 쓰는 스트림.
|
||||||
|
///
|
||||||
|
/// **왜 필요한가.** 이클립스 집계는 HOME 배너·GALAXY 배너·행성 HUD·방어 화면이
|
||||||
|
/// 동시에 본다. 화면마다 `snapshots()` 를 열면 같은 문서에 리스너가 화면 수만큼
|
||||||
|
/// 붙고, 화면이 겹쳐 뜨면(탭 위에 모달) 더 늘어난다. 실측에서 한 팬덤에 리스너가
|
||||||
|
/// 최대 7 건까지 붙었다. 읽기 비용이 그만큼 늘고, 값이 도착하는 순서도 갈린다.
|
||||||
|
///
|
||||||
|
/// 여기서는 첫 구독자가 원본을 열고 마지막 구독자가 떠날 때 닫는다. 그 사이의
|
||||||
|
/// 구독자는 모두 같은 원본을 나눠 본다.
|
||||||
|
///
|
||||||
|
/// **새 구독자는 마지막 값을 곧바로 받는다.** 그러지 않으면 나중에 열린 화면이
|
||||||
|
/// 다음 갱신이 올 때까지 빈 상태로 남는다 — 이클립스는 몇 분에 한 번 바뀌므로
|
||||||
|
/// 그 빈 상태가 오래간다.
|
||||||
|
class SharedStream<T> {
|
||||||
|
SharedStream(this._open);
|
||||||
|
|
||||||
|
/// 원본을 여는 함수. 실제로 구독자가 생겼을 때만 불린다.
|
||||||
|
final Stream<T> Function() _open;
|
||||||
|
|
||||||
|
StreamSubscription<T>? _sub;
|
||||||
|
final _out = StreamController<T>.broadcast();
|
||||||
|
|
||||||
|
T? _last;
|
||||||
|
bool _hasLast = false;
|
||||||
|
|
||||||
|
/// 구독자 수. 0 이 되면 원본을 닫는다.
|
||||||
|
int _listeners = 0;
|
||||||
|
|
||||||
|
Stream<T> get stream => Stream<T>.multi((controller) {
|
||||||
|
_listeners += 1;
|
||||||
|
_openIfNeeded();
|
||||||
|
// 늦게 붙은 구독자에게 현재 값을 먼저 준다.
|
||||||
|
if (_hasLast) controller.add(_last as T);
|
||||||
|
final sub = _out.stream.listen(
|
||||||
|
controller.add,
|
||||||
|
onError: controller.addError,
|
||||||
|
onDone: controller.close,
|
||||||
|
);
|
||||||
|
controller.onCancel = () async {
|
||||||
|
await sub.cancel();
|
||||||
|
_listeners -= 1;
|
||||||
|
if (_listeners == 0) await _close();
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
|
void _openIfNeeded() {
|
||||||
|
if (_sub != null) return;
|
||||||
|
_sub = _open().listen(
|
||||||
|
(value) {
|
||||||
|
_last = value;
|
||||||
|
_hasLast = true;
|
||||||
|
if (!_out.isClosed) _out.add(value);
|
||||||
|
},
|
||||||
|
onError: (Object error, StackTrace stack) {
|
||||||
|
if (!_out.isClosed) _out.addError(error, stack);
|
||||||
|
},
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
Future<void> _close() async {
|
||||||
|
await _sub?.cancel();
|
||||||
|
_sub = null;
|
||||||
|
// 마지막 값은 버린다. 다시 열릴 때 옛 값을 먼저 주면 화면이 잠깐 과거를 그린다.
|
||||||
|
_last = null;
|
||||||
|
_hasLast = false;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 저장소가 버려질 때 부른다.
|
||||||
|
Future<void> dispose() async {
|
||||||
|
await _close();
|
||||||
|
await _out.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,76 @@
|
|||||||
|
import 'dart:async';
|
||||||
|
|
||||||
|
import 'package:fancake/core/data/shared_stream.dart';
|
||||||
|
import 'package:flutter_test/flutter_test.dart';
|
||||||
|
|
||||||
|
/// 실시간 구독 공유 — 같은 문서를 여러 화면이 볼 때 원본 구독은 하나만 연다.
|
||||||
|
void main() {
|
||||||
|
group('구독 공유 — 여러 화면이 원본 하나를 나눠 본다', () {
|
||||||
|
test('여러 화면이 같은 원본 하나를 나눠 쓴다', () async {
|
||||||
|
// 화면마다 snapshots() 를 열면 한 팬덤에 리스너가 화면 수만큼 붙는다.
|
||||||
|
var opened = 0;
|
||||||
|
final controller = StreamController<int>.broadcast();
|
||||||
|
addTearDown(controller.close);
|
||||||
|
final shared = SharedStream<int>(() {
|
||||||
|
opened += 1;
|
||||||
|
return controller.stream;
|
||||||
|
});
|
||||||
|
|
||||||
|
final a = <int>[];
|
||||||
|
final b = <int>[];
|
||||||
|
final subA = shared.stream.listen(a.add);
|
||||||
|
final subB = shared.stream.listen(b.add);
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
|
||||||
|
controller.add(7);
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
|
||||||
|
// 원본은 한 번만 열렸는데 두 화면 모두 값을 받았다.
|
||||||
|
expect(opened, 1);
|
||||||
|
expect(a, [7]);
|
||||||
|
expect(b, [7]);
|
||||||
|
|
||||||
|
await subA.cancel();
|
||||||
|
await subB.cancel();
|
||||||
|
});
|
||||||
|
|
||||||
|
test('늦게 붙은 화면이 현재 값을 곧바로 받는다', () async {
|
||||||
|
final controller = StreamController<int>.broadcast();
|
||||||
|
addTearDown(controller.close);
|
||||||
|
final shared = SharedStream<int>(() => controller.stream);
|
||||||
|
|
||||||
|
final first = <int>[];
|
||||||
|
final subA = shared.stream.listen(first.add);
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
controller.add(42);
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
|
||||||
|
// 뒤늦게 열린 화면이 다음 갱신까지 빈 상태로 남으면 안 된다.
|
||||||
|
final late = <int>[];
|
||||||
|
final subB = shared.stream.listen(late.add);
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
expect(late, [42]);
|
||||||
|
|
||||||
|
await subA.cancel();
|
||||||
|
await subB.cancel();
|
||||||
|
});
|
||||||
|
|
||||||
|
test('마지막 화면이 떠나면 원본을 닫는다', () async {
|
||||||
|
var cancelled = false;
|
||||||
|
final controller = StreamController<int>(
|
||||||
|
onCancel: () => cancelled = true,
|
||||||
|
);
|
||||||
|
addTearDown(controller.close);
|
||||||
|
final shared = SharedStream<int>(() => controller.stream);
|
||||||
|
|
||||||
|
final sub = shared.stream.listen((_) {});
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
expect(cancelled, isFalse);
|
||||||
|
|
||||||
|
await sub.cancel();
|
||||||
|
await Future<void>.delayed(Duration.zero);
|
||||||
|
// 남겨 두면 화면이 없는 동안에도 읽기 비용이 계속 든다.
|
||||||
|
expect(cancelled, isTrue);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user