fixup! feat: implement exploreDirectory method

eclipse-thingweb · Dec 22, 2023 · 5b5871e · 5b5871e
1 parent 5eabcab
commit 5b5871e
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 5 deletions.
diff --git a/lib/src/core/thing_discovery.dart b/lib/src/core/thing_discovery.dart
@@ -414,17 +414,18 @@ class ThingDiscoveryProcess extends Stream<ThingDescription>
   }) {
     final streamSubscription = _thingDescriptionStream.listen(
       onData,
-      onError: (error) {
+      onError: (error, stackTrace) {
         if (error is Exception) {
           _error = error;
+          // ignore: avoid_dynamic_calls
+          onError?.call(error, stackTrace);
         }
-        // ignore: avoid_dynamic_calls
-        onError?.call(error);
       },
-      onDone: () async {
+      onDone: () {
         _done = true;
         onDone?.call();
       },
+      cancelOnError: cancelOnError,
     );
 
     _streamSubscription = streamSubscription;

diff --git a/lib/src/core/wot.dart b/lib/src/core/wot.dart
@@ -4,6 +4,8 @@
 //
 // SPDX-License-Identifier: BSD-3-Clause
 
+import 'dart:async';
+
 import '../../scripting_api.dart' as scripting_api;
 import '../definitions/thing_description.dart';
 import '../scripting_api/discovery/discovery_method.dart';
@@ -125,7 +127,33 @@ class WoT implements scripting_api.WoT {
 
     final thingDescriptionStream = Stream.fromIterable(
       rawThingDescriptions.whereType<Map<String, dynamic>>(),
-    ).map(ThingDescription.fromJson);
+    ).transform(
+      StreamTransformer<Map<String, dynamic>, ThingDescription>(
+        (stream, cancelOnError) {
+          final streamController = StreamController<ThingDescription>();
+
+          final streamSubscription = stream.listen(
+            (rawThingDescription) {
+              try {
+                streamController
+                    .add(ThingDescription.fromJson(rawThingDescription));
+              } on Exception catch (exception) {
+                streamController.addError(exception);
+              }
+            },
+            onDone: streamController.close,
+            onError: streamController.addError,
+            cancelOnError: cancelOnError,
+          );
+          streamController
+            ..onPause = streamSubscription.pause
+            ..onResume = streamSubscription.resume
+            ..onCancel = streamSubscription.cancel;
+
+          return streamController.stream.listen(null);
+        },
+      ),
+    );
 
     return ThingDiscoveryProcess(thingDescriptionStream, filter);
   }