Use a StreamController in onStreamedFrameAvailable

We need to have exception-throwing `onPause` and `onResume` callback,
as they are required to throw by the camera platform interface.

packages/camera/camera_windows/lib/camera_windows.dart

@@ -237,18 +237,30 @@ class CameraWindows extends CameraPlatform {
 
   @override
   Stream<CameraImageData> onStreamedFrameAvailable(int cameraId,
-      {CameraImageStreamOptions? options}) async* {
-    await _hostApi.startImageStream(cameraId);
-    final eventChannelName =
-        'plugins.flutter.io/camera_windows/imageStream/$cameraId';
-    final EventChannel imageStreamChannel = EventChannel(eventChannelName);
-    try {
-      await for (final frame in imageStreamChannel.receiveBroadcastStream()) {
-        yield cameraImageFromPlatformData(frame as Map<dynamic, dynamic>);
-      }
-    } finally {
-      await _hostApi.stopImageStream(cameraId);
-    }
+      {CameraImageStreamOptions? options}) {
+    late StreamController<CameraImageData> controller;
+
+    controller = StreamController<CameraImageData>(
+        onListen: () async {
+          final eventChannelName = await _hostApi.startImageStream(cameraId);
+          final EventChannel imageStreamChannel =
+              EventChannel(eventChannelName);
+          imageStreamChannel.receiveBroadcastStream().listen((dynamic image) =>
+              controller.add(
+                  cameraImageFromPlatformData(image as Map<dynamic, dynamic>)));
+        },
+        onPause: _onFrameStreamPauseResume,
+        onResume: _onFrameStreamPauseResume,
+        onCancel: () async {
+          await _hostApi.stopImageStream(cameraId);
+        });
+
+    return controller.stream;
+  }
+
+  void _onFrameStreamPauseResume() {
+    throw CameraException('InvalidCall',
+        'Pause and resume are not supported for onStreamedFrameAvailable');
   }
 
   @override

packages/camera/camera_windows/lib/src/messages.g.dart

@@ -363,7 +363,8 @@ class CameraApi {
   }
 
   /// Starts the image stream for the given camera.
-  Future<void> startImageStream(int cameraId) async {
+  /// Returns the name of the [EventChannel] used to deliver the images.
+  Future<String> startImageStream(int cameraId) async {
     final String pigeonVar_channelName =
         'dev.flutter.pigeon.camera_windows.CameraApi.startImageStream$pigeonVar_messageChannelSuffix';
     final BasicMessageChannel<Object?> pigeonVar_channel =
@@ -382,8 +383,13 @@ class CameraApi {
         message: pigeonVar_replyList[1] as String?,
         details: pigeonVar_replyList[2],
       );
+    } else if (pigeonVar_replyList[0] == null) {
+      throw PlatformException(
+        code: 'null-error',
+        message: 'Host platform returned null value for non-null return value.',
+      );
     } else {
-      return;
+      return (pigeonVar_replyList[0] as String?)!;
     }
   }
 

packages/camera/camera_windows/pigeons/messages.dart

@@ -71,7 +71,8 @@ abstract class CameraApi {
   String stopVideoRecording(int cameraId);
 
   /// Starts the image stream for the given camera.
-  void startImageStream(int cameraId);
+  /// Returns the name of the [EventChannel] used to deliver the images.
+  String startImageStream(int cameraId);
 
   /// Stops the image stream for the given camera.
   void stopImageStream(int cameraId);

packages/camera/camera_windows/windows/camera_plugin.cpp

@@ -346,7 +346,7 @@ void CameraPlugin::StopVideoRecording(
   }
 }
 
-std::optional<FlutterError> CameraPlugin::StartImageStream(int64_t camera_id) {
+ErrorOr<std::string> CameraPlugin::StartImageStream(int64_t camera_id) {
   Camera* camera = GetCameraByCameraId(camera_id);
   if (!camera) {
     return FlutterError("camera_error", "Camera not created");
@@ -355,6 +355,11 @@ std::optional<FlutterError> CameraPlugin::StartImageStream(int64_t camera_id) {
   CaptureController* cc = camera->GetCaptureController();
   assert(cc);
 
+  if (cc->IsStreaming()) {
+    return FlutterError("camera_error",
+                        "Images from camera are already streaming");
+  }
+
   std::ostringstream event_channel_name;
   event_channel_name << "plugins.flutter.io/camera_windows/imageStream/"
                      << camera_id;
@@ -376,7 +381,7 @@ std::optional<FlutterError> CameraPlugin::StartImageStream(int64_t camera_id) {
 
   frame_event_channel.SetStreamHandler(std::move(event_channel_handler));
 
-  return std::nullopt;
+  return event_channel_name.str();
 }
 
 std::optional<FlutterError> CameraPlugin::StopImageStream(int64_t camera_id) {

packages/camera/camera_windows/windows/camera_plugin.h

@@ -69,7 +69,7 @@ class CameraPlugin : public flutter::Plugin,
   void StopVideoRecording(
       int64_t camera_id,
       std::function<void(ErrorOr<std::string> reply)> result) override;
-  std::optional<FlutterError> StartImageStream(int64_t camera_id) override;
+  ErrorOr<std::string> StartImageStream(int64_t camera_id) override;
   std::optional<FlutterError> StopImageStream(int64_t camera_id) override;
   void TakePicture(
       int64_t camera_id,

packages/camera/camera_windows/windows/capture_controller.h

@@ -104,6 +104,8 @@ class CaptureController {
   // Stops the current image streaming.
   virtual void StopImageStream() = 0;
 
+  virtual bool IsStreaming() const = 0;
+
   // Captures a still photo.
   virtual void TakePicture(const std::string& file_path) = 0;
 };
@@ -142,6 +144,9 @@ class CaptureControllerImpl : public CaptureController,
       std::unique_ptr<flutter::EventSink<flutter::EncodableValue>> sink)
       override;
   void StopImageStream() override;
+  bool IsStreaming() const override {
+    return static_cast<bool>(image_stream_sink_);
+  }
   void TakePicture(const std::string& file_path) override;
 
   // CaptureEngineObserver

packages/camera/camera_windows/windows/messages.g.cpp

@@ -516,14 +516,14 @@ void CameraApi::SetUp(flutter::BinaryMessenger* binary_messenger,
                 return;
               }
               const int64_t camera_id_arg = encodable_camera_id_arg.LongValue();
-              std::optional<FlutterError> output =
+              ErrorOr<std::string> output =
                   api->StartImageStream(camera_id_arg);
-              if (output.has_value()) {
-                reply(WrapError(output.value()));
+              if (output.has_error()) {
+                reply(WrapError(output.error()));
                 return;
               }
               EncodableList wrapped;
-              wrapped.push_back(EncodableValue());
+              wrapped.push_back(EncodableValue(std::move(output).TakeValue()));
               reply(EncodableValue(std::move(wrapped)));
             } catch (const std::exception& exception) {
               reply(WrapError(exception.what()));

packages/camera/camera_windows/windows/messages.g.h

@@ -190,7 +190,8 @@ class CameraApi {
       int64_t camera_id,
       std::function<void(ErrorOr<std::string> reply)> result) = 0;
   // Starts the image stream for the given camera.
-  virtual std::optional<FlutterError> StartImageStream(int64_t camera_id) = 0;
+  // Returns the name of the [EventChannel] used to deliver the images.
+  virtual ErrorOr<std::string> StartImageStream(int64_t camera_id) = 0;
   // Stops the image stream for the given camera.
   virtual std::optional<FlutterError> StopImageStream(int64_t camera_id) = 0;
   // Starts the preview stream for the given camera.

packages/camera/camera_windows/windows/test/camera_plugin_test.cpp

@@ -363,6 +363,11 @@ TEST(CameraPlugin, StartImageStreamHandlerCallsStartImageStream) {
   std::unique_ptr<MockCaptureController> capture_controller =
       std::make_unique<MockCaptureController>();
 
+  std::unique_ptr<MockTextureRegistrar> texture_registrar =
+      std::make_unique<MockTextureRegistrar>();
+  std::unique_ptr<MockBinaryMessenger> messenger =
+      std::make_unique<MockBinaryMessenger>();
+
   EXPECT_CALL(*camera, HasCameraId(Eq(mock_camera_id)))
       .Times(1)
       .WillOnce([cam = camera.get()](int64_t camera_id) {
@@ -374,19 +379,20 @@ TEST(CameraPlugin, StartImageStreamHandlerCallsStartImageStream) {
       .WillOnce(
           [cam = camera.get()]() { return cam->capture_controller_.get(); });
 
-  EXPECT_CALL(*capture_controller, StartImageStream).Times(1);
+  EXPECT_CALL(*capture_controller, IsStreaming).WillRepeatedly(Return(false));
+
+  EXPECT_CALL(*messenger, SetMessageHandler).Times(1);
 
   camera->camera_id_ = mock_camera_id;
   camera->capture_controller_ = std::move(capture_controller);
 
-  MockCameraPlugin plugin(std::make_unique<MockTextureRegistrar>().get(),
-                          std::make_unique<MockBinaryMessenger>().get(),
+  MockCameraPlugin plugin(texture_registrar.get(), messenger.get(),
                           std::make_unique<MockCameraFactory>());
 
   // Add mocked camera to plugins camera list.
   plugin.AddCamera(std::move(camera));
 
-  EXPECT_EQ(plugin.StartImageStream(mock_camera_id), std::nullopt);
+  EXPECT_FALSE(plugin.StartImageStream(mock_camera_id).has_error());
 }
 
 TEST(CameraPlugin, StartImageStreamHandlerErrorOnInvalidCameraId) {
@@ -419,8 +425,8 @@ TEST(CameraPlugin, StartImageStreamHandlerErrorOnInvalidCameraId) {
   // Add mocked camera to plugins camera list.
   plugin.AddCamera(std::move(camera));
 
-  EXPECT_THAT(plugin.StartImageStream(missing_camera_id),
-              Optional(Property("code", &FlutterError::code, "camera_error")));
+  auto result = plugin.StartImageStream(missing_camera_id);
+  EXPECT_TRUE(result.has_error());
 }
 
 TEST(CameraPlugin, StopImageStreamHandlerCallsStopImageStream) {

packages/camera/camera_windows/windows/test/mocks.h

@@ -255,6 +255,7 @@ class MockCaptureController : public CaptureController {
       (std::unique_ptr<flutter::EventSink<flutter::EncodableValue>> sink),
       (override));
   MOCK_METHOD(void, StopImageStream, (), (override));
+  MOCK_METHOD(bool, IsStreaming, (), (const override));
   MOCK_METHOD(void, TakePicture, (const std::string& file_path), (override));
 };