stream packets

* use ports for objc

* add task reaper

* cleanup, fewer allocations

* link to pr

mobile/lib/domain/services/timeline.service.dart

@@ -85,12 +85,8 @@ class TimelineFactory {
   TimelineService fromAssetsWithBuckets(List<BaseAsset> assets, TimelineOrigin type) =>
       TimelineService(_timelineRepository.fromAssetsWithBuckets(assets, type));
 
-  /// Creates a TimelineService for serving geographical map queries, such assets within bounded locations
-  TimelineService geographicMap(
-    List<String> userIds,
-    TimelineMapOptions Function() currentOptions,
-    Stream<TimelineMapOptions> optionsStream,
-  ) => TimelineService(_timelineRepository.geographicMap(userIds, currentOptions, optionsStream, groupBy));
+  TimelineService map(List<String> userIds, TimelineMapOptions options) =>
+      TimelineService(_timelineRepository.map(userIds, options, groupBy));
 }
 
 class TimelineService {

mobile/lib/infrastructure/repositories/timeline.repository.dart

@@ -509,21 +509,9 @@ class DriftTimelineRepository extends DriftDatabaseRepository {
     return query.map((row) => row.toDto()).get();
   }
 
-  /// Creates a geographic map query that can dynamically filter on changing [TimelineMapOptions]
-  /// (most notably the active map bounds)
-  TimelineQuery geographicMap(
-    List<String> userIds,
-    TimelineMapOptions Function() currentOptions,
-    Stream<TimelineMapOptions> optionsStream,
-    GroupAssetsBy groupBy,
-  ) => (
-    bucketSource: () => Stream.value(currentOptions())
-        .followedBy(optionsStream)
-        .switchMap(
-          // Any error would kill the stream for all options; make sure the stream stays alive
-          (options) => _watchMapBucket(userIds, options, groupBy: groupBy).handleError((_) {}),
-        ),
-    assetSource: (offset, count) => _getMapBucketAssets(userIds, currentOptions(), offset: offset, count: count),
+  TimelineQuery map(List<String> userIds, TimelineMapOptions options, GroupAssetsBy groupBy) => (
+    bucketSource: () => _watchMapBucket(userIds, options, groupBy: groupBy),
+    assetSource: (offset, count) => _getMapBucketAssets(userIds, options, offset: offset, count: count),
     origin: TimelineOrigin.map,
   );
 

mobile/lib/presentation/widgets/bottom_sheet/map_bottom_sheet.widget.dart

@@ -35,6 +35,7 @@ class _ScopedMapTimeline extends StatelessWidget {
 
   @override
   Widget build(BuildContext context) {
+    // TODO: this causes the timeline to switch to flicker to "loading" state and back. This is both janky and inefficient.
     return ProviderScope(
       overrides: [
         timelineServiceProvider.overrideWith((ref) {
@@ -43,16 +44,13 @@ class _ScopedMapTimeline extends StatelessWidget {
             throw Exception('User must be logged in to access archive');
           }
 
-          final withPartners = ref.watch(mapStateProvider.select((s) => s.withPartners));
-          final users = withPartners ? ref.watch(timelineUsersProvider).valueOrNull ?? [user.id] : [user.id];
+          final users = ref.watch(mapStateProvider).withPartners
+              ? ref.watch(timelineUsersProvider).valueOrNull ?? [user.id]
+              : [user.id];
 
           final timelineService = ref
               .watch(timelineFactoryProvider)
-              .geographicMap(
-                users,
-                () => ref.read(mapStateProvider).toOptions(),
-                ref.read(mapStateProvider.notifier).optionsStream,
-              );
+              .map(users, ref.watch(mapStateProvider).toOptions());
           ref.onDispose(timelineService.dispose);
           return timelineService;
         }),

mobile/lib/presentation/widgets/map/map.state.dart

@@ -1,5 +1,3 @@
-import 'dart:async';
-
 import 'package:flutter/material.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/domain/models/events.model.dart';
@@ -65,16 +63,11 @@ class MapState {
 class MapStateNotifier extends Notifier<MapState> {
   MapStateNotifier();
 
-  final StreamController<TimelineMapOptions> _optionsController = StreamController.broadcast();
-
-  Stream<TimelineMapOptions> get optionsStream => _optionsController.stream;
-
   bool setBounds(LatLngBounds bounds) {
     if (state.bounds == bounds) {
       return false;
     }
     state = state.copyWith(bounds: bounds);
-    _optionsController.add(state.toOptions());
     return true;
   }
 
@@ -89,14 +82,12 @@ class MapStateNotifier extends Notifier<MapState> {
   void switchFavoriteOnly(bool isFavoriteOnly) {
     ref.read(settingsProvider).write(.mapShowFavoriteOnly, isFavoriteOnly);
     state = state.copyWith(onlyFavorites: isFavoriteOnly);
-    _optionsController.add(state.toOptions());
     EventStream.shared.emit(const MapMarkerReloadEvent());
   }
 
   void switchIncludeArchived(bool isIncludeArchived) {
     ref.read(settingsProvider).write(.mapIncludeArchived, isIncludeArchived);
     state = state.copyWith(includeArchived: isIncludeArchived);
-    _optionsController.add(state.toOptions());
     EventStream.shared.emit(const MapMarkerReloadEvent());
   }
 
@@ -109,13 +100,11 @@ class MapStateNotifier extends Notifier<MapState> {
   void setRelativeTime(int relativeDays) {
     ref.read(settingsProvider).write(.mapRelativeDate, relativeDays);
     state = state.copyWith(relativeDays: relativeDays);
-    _optionsController.add(state.toOptions());
     EventStream.shared.emit(const MapMarkerReloadEvent());
   }
 
   @override
   MapState build() {
-    ref.onDispose(_optionsController.close);
     final mapConfig = ref.read(appConfigProvider.select((config) => config.map));
     return MapState(
       themeMode: mapConfig.themeMode,

mobile/pubspec.lock

@@ -309,8 +309,8 @@ packages:
     dependency: "direct main"
     description:
       path: "pkgs/cupertino_http"
-      ref: a0a933358517c6d01cff37fc2a2752ee2d744a3c
-      resolved-ref: a0a933358517c6d01cff37fc2a2752ee2d744a3c
+      ref: "58b03c756b81d16b3975a8ae4b91d25360123bb5"
+      resolved-ref: "58b03c756b81d16b3975a8ae4b91d25360123bb5"
       url: "https://github.com/mertalev/http"
     source: git
     version: "3.0.0-wip"
@@ -1158,12 +1158,13 @@ packages:
     source: hosted
     version: "0.5.0"
   objective_c:
-    dependency: transitive
+    dependency: "direct overridden"
     description:
-      name: objective_c
-      sha256: "6cb691c686fa2838c6deb34980d426145c2a5d537491cb83d463c33cdbc726ed"
-      url: "https://pub.dev"
-    source: hosted
+      path: "pkgs/objective_c"
+      ref: "2915556701f4734a784222f290a82adb1b101682"
+      resolved-ref: "2915556701f4734a784222f290a82adb1b101682"
+      url: "https://github.com/mertalev/native"
+    source: git
     version: "9.4.1"
   octo_image:
     dependency: "direct main"

mobile/pubspec.yaml

@@ -84,7 +84,7 @@ dependencies:
   cupertino_http:
     git:
       url: https://github.com/mertalev/http
-      ref: 'a0a933358517c6d01cff37fc2a2752ee2d744a3c' # https://github.com/dart-lang/http/pull/1876
+      ref: '58b03c756b81d16b3975a8ae4b91d25360123bb5'
       path: pkgs/cupertino_http/
   ok_http:
     git:
@@ -114,6 +114,11 @@ dev_dependencies:
 # Pin bonsoir to 5.x until cast releases a version compatible with bonsoir 6.x.
 dependency_overrides:
   bonsoir: ^5.1.11
+  objective_c:
+    git:
+      url: https://github.com/mertalev/native
+      ref: '2915556701f4734a784222f290a82adb1b101682' # https://github.com/dart-lang/native/pull/3458
+      path: pkgs/objective_c/
 
 flutter:
   uses-material-design: true

server/src/repositories/media.repository.ts

@@ -3,10 +3,9 @@ import { ExifDateTime, exiftool, WriteTags } from 'exiftool-vendored';
 import ffmpeg, { FfprobeData, FfprobeStream } from 'fluent-ffmpeg';
 import _ from 'lodash';
 import { Duration } from 'luxon';
-import { execFile as execFileCb } from 'node:child_process';
+import { spawn } from 'node:child_process';
 import fs from 'node:fs/promises';
 import { Writable } from 'node:stream';
-import { promisify } from 'node:util';
 import sharp from 'sharp';
 import { ORIENTATION_TO_SHARP_ROTATION } from 'src/constants';
 import { Exif } from 'src/database';
@@ -44,8 +43,6 @@ const probe = (input: string, options: string[]): Promise<FfprobeData> =>
     ffmpeg.ffprobe(input, options, (error, data) => (error ? reject(error) : resolve(data))),
   );
 
-const execFile = promisify(execFileCb);
-
 sharp.concurrency(0);
 sharp.cache({ files: 0 });
 
@@ -291,33 +288,37 @@ export class MediaRepository {
    * Needed for accurate segments, especially when remuxing, seeking and/or VFR is involved.
    * Scanning packets for keyframes in JS is much faster than -skip_frame nokey since it avoids decoding the video.
    */
-  async probePackets(input: string, streamIndex: number): Promise<VideoPacketInfo | null> {
-    const { stdout } = await execFile('ffprobe', [
-      '-v',
-      'error',
-      '-select_streams',
-      String(streamIndex),
-      '-show_entries',
-      'packet=pts,duration,flags',
-      '-of',
-      'csv=p=0',
-      input,
-    ]);
+  probePackets(input: string, streamIndex: number): Promise<VideoPacketInfo | null> {
+    const ffprobe = spawn(
+      'ffprobe',
+      [
+        '-v',
+        'error',
+        '-select_streams',
+        String(streamIndex),
+        '-show_entries',
+        'packet=pts,duration,flags',
+        '-of',
+        'csv=p=0',
+        input,
+      ],
+      { stdio: ['ignore', 'pipe', 'pipe'] },
+    );
 
     let totalDuration = 0;
     const keyframePts: number[] = [];
     const keyframeAccDuration: number[] = [];
     const keyframeOwnDuration: number[] = [];
     const postDiscard: { pts: number; duration: number }[] = [];
-    for (const line of stdout.split('\n')) {
+    const parseLine = (line: string) => {
       if (!line) {
-        continue;
+        return;
       }
       const [ptsStr, durationStr, flags] = line.split(',');
       const pts = Number.parseInt(ptsStr);
       const duration = Number.parseInt(durationStr);
-      if (Number.isNaN(pts) || Number.isNaN(duration)) {
-        continue;
+      if (Number.isNaN(pts) || Number.isNaN(duration) || !flags) {
+        return;
       }
       // Discarded packets don't contribute to packet count, but still contribute to video duration
       totalDuration += duration;
@@ -332,20 +333,43 @@ export class MediaRepository {
         // Non-keyframes are accounted for in totalDuration.
         keyframeOwnDuration.push(duration);
       }
-    }
-
-    if (postDiscard.length === 0) {
-      return null;
-    }
-
-    return {
-      totalDuration,
-      packetCount: postDiscard.length,
-      outputFrames: this.cfrOutputFrames(postDiscard, postDiscard.length / totalDuration),
-      keyframePts,
-      keyframeAccDuration,
-      keyframeOwnDuration,
     };
+
+    let stderr = '';
+    let remainder = '';
+    ffprobe.stderr.setEncoding('utf8');
+    ffprobe.stderr.on('data', (chunk: string) => (stderr += chunk));
+    ffprobe.stdout.setEncoding('utf8');
+    ffprobe.stdout.on('data', (chunk: string) => {
+      const lines = chunk.split('\n');
+      lines[0] = remainder + lines[0];
+      remainder = lines.pop() as string;
+      for (const line of lines) {
+        parseLine(line);
+      }
+    });
+
+    return new Promise<VideoPacketInfo | null>((resolve, reject) => {
+      ffprobe.on('error', reject);
+      ffprobe.on('close', (code) => {
+        if (code !== 0) {
+          return reject(new Error(`ffprobe exited with code ${code}: ${stderr.trim()}`));
+        }
+        parseLine(remainder);
+        if (postDiscard.length === 0) {
+          return resolve(null);
+        }
+
+        resolve({
+          totalDuration,
+          packetCount: postDiscard.length,
+          outputFrames: this.cfrOutputFrames(postDiscard, postDiscard.length / totalDuration),
+          keyframePts,
+          keyframeAccDuration,
+          keyframeOwnDuration,
+        });
+      });
+    });
   }
 
   transcode(input: string, output: string | Writable, options: TranscodeCommand): Promise<void> {