Merge pull request #204 from eclipse-thingweb/fix-tdd-discovery

Make directory discovery a bit more robust

example/tdd_discovery.dart

@@ -0,0 +1,30 @@
+// Copyright 2024 Contributors to the Eclipse Foundation. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+//
+// SPDX-License-Identifier: BSD-3-Clause
+
+// ignore_for_file: avoid_print
+
+import "package:dart_wot/binding_http.dart";
+import "package:dart_wot/core.dart";
+
+Future<void> main() async {
+  final servient = Servient.create(
+    clientFactories: [
+      HttpClientFactory(),
+    ],
+  );
+  final wot = await servient.start();
+
+  final url = Uri.parse("http://plugfest.thingweb.io:8081");
+  print("Requesting TD from $url ...");
+  final thingDiscoveryProcess = await wot.exploreDirectory(url);
+
+  thingDiscoveryProcess.listen(
+    (thingDescription) => print(thingDescription.title),
+    onError: print,
+  );
+
+  await servient.shutdown();
+}

lib/src/core/definitions/extensions/json_parser.dart

@@ -61,7 +61,9 @@ extension ParseField on Map<String, dynamic> {
       return <String, dynamic>{} as T;
     }
 
-    throw FormatException("Expected $T, got ${fieldValue.runtimeType}");
+    throw FormatException(
+      "Expected $T, got ${fieldValue.runtimeType} for field $name",
+    );
   }
 
   /// Parses a single field with a given [name] as a [Uri].

lib/src/core/implementation/servient.dart

@@ -4,6 +4,8 @@
 //
 // SPDX-License-Identifier: BSD-3-Clause
 
+import "dart:developer";
+
 import "package:meta/meta.dart";
 import "package:uuid/uuid.dart";
 
@@ -379,6 +381,21 @@ class InternalServient implements Servient {
     return ThingDescription.fromJson(dataSchemaValue.value);
   }
 
+  Stream<ThingDescription> _processThingDescriptions(
+    List<Object?> rawThingDescriptions,
+  ) async* {
+    for (final rawThingDescription in rawThingDescriptions) {
+      if (rawThingDescription is Map<String, Object?>) {
+        try {
+          yield rawThingDescription.toThingDescription();
+        } on Exception catch (e) {
+          log(e.toString());
+          yield* Stream.error(e);
+        }
+      }
+    }
+  }
+
   /// Retrieves [ThingDescription] from a Thing Description Directory (TDD).
   ///
   /// This method expects the TDD's Thing Description to be located under the
@@ -435,9 +452,8 @@ class InternalServient implements Servient {
       );
     }
 
-    final thingDescriptionStream = Stream.fromIterable(
-      rawThingDescriptions.whereType<Map<String, Object?>>(),
-    ).map((rawThingDescription) => rawThingDescription.toThingDescription());
+    final thingDescriptionStream =
+        _processThingDescriptions(rawThingDescriptions);
 
     return ThingDiscoveryProcess(thingDescriptionStream, thingFilter);
   }

lib/src/core/implementation/thing_discovery.dart

@@ -485,11 +485,11 @@ class ThingDiscoveryProcess extends Stream<ThingDescription>
   }) {
     final streamSubscription = _thingDescriptionStream.listen(
       onData,
-      onError: (error, stackTrace) {
+      onError: (error) {
         if (error is Exception) {
           _error = error;
           // ignore: avoid_dynamic_calls
-          onError?.call(error, stackTrace);
+          onError?.call(error);
         }
       },
       onDone: () {

test/core/discovery_test.dart

@@ -352,7 +352,7 @@ void main() {
         (event) {
           counter++;
         },
-        onError: (error, stackTrace) async {},
+        onError: (error) async {},
         onDone: () {
           expect(counter, 0);
           expect(thingDiscoveryProcess.done, true);