feat!: rework discoverer APIs

Author: JKRhb

lib/src/binding_coap/coap_client.dart

@@ -73,7 +73,8 @@ coap.PskCredentialsCallback? _createPskCallback(
 }
 
 /// A [ProtocolClient] for the Constrained Application Protocol (CoAP).
-final class CoapClient extends ProtocolClient {
+final class CoapClient extends ProtocolClient
+    with DirectDiscoverer, MulticastDiscoverer, CoreLinkFormatDiscoverer {
   /// Creates a new [CoapClient] based on an optional [CoapConfig].
   CoapClient({
     CoapConfig? coapConfig,
@@ -446,59 +447,14 @@ final class CoapClient extends ProtocolClient {
   @override
   Future<void> stop() async {}
 
-  Stream<DiscoveryContent> _discoverFromMulticast(
-    coap.CoapClient client,
-    Uri uri,
-  ) async* {
-    final streamController = StreamController<DiscoveryContent>();
-    final multicastResponseHandler = coap.CoapMulticastResponseHandler(
-      (data) {
-        streamController.add(data.determineDiscoveryContent(uri.scheme));
-      },
-      onError: streamController.addError,
-      onDone: () async {
-        await streamController.close();
-      },
-    );
-
-    final content = _sendDiscoveryRequest(
-      uri,
-      coap.RequestMethod.get,
-      form: null,
-      accept: coap.CoapMediaType.applicationTdJson,
-      multicastResponseHandler: multicastResponseHandler,
-    );
-    unawaited(content);
-    yield* streamController.stream;
-  }
-
-  Stream<DiscoveryContent> _discoverFromUnicast(
-    coap.CoapClient client,
-    Uri uri,
-  ) async* {
-    yield await _sendDiscoveryRequest(
-      uri,
-      coap.RequestMethod.get,
-      form: null,
-      accept: coap.CoapMediaType.applicationTdJson,
-    );
-  }
-
   @override
-  Stream<DiscoveryContent> discoverDirectly(
-    Uri uri, {
-    bool disableMulticast = false,
-  }) async* {
-    final client = coap.CoapClient(uri);
-
-    if (uri.isMulticastAddress) {
-      if (!disableMulticast) {
-        yield* _discoverFromMulticast(client, uri);
-      }
-    } else {
-      yield* _discoverFromUnicast(client, uri);
-    }
-  }
+  Future<DiscoveryContent> discoverDirectly(Uri uri) async =>
+      _sendDiscoveryRequest(
+        uri,
+        coap.RequestMethod.get,
+        form: null,
+        accept: coap.CoapMediaType.applicationTdJson,
+      );
 
   @override
   Stream<DiscoveryContent> discoverWithCoreLinkFormat(Uri uri) async* {
@@ -534,10 +490,26 @@ final class CoapClient extends ProtocolClient {
   }
 
   @override
-  Future<Content> requestThingDescription(Uri url) async => _sendRequest(
-        url,
-        coap.RequestMethod.get,
-        form: null,
-        accept: coap.CoapMediaType.applicationTdJson,
-      );
+  Stream<Content> discoverViaMulticast(Uri uri) async* {
+    final streamController = StreamController<DiscoveryContent>();
+    final multicastResponseHandler = coap.CoapMulticastResponseHandler(
+      (data) {
+        streamController.add(data.determineDiscoveryContent(uri.scheme));
+      },
+      onError: streamController.addError,
+      onDone: () async {
+        await streamController.close();
+      },
+    );
+
+    final content = _sendDiscoveryRequest(
+      uri,
+      coap.RequestMethod.get,
+      form: null,
+      accept: coap.CoapMediaType.applicationTdJson,
+      multicastResponseHandler: multicastResponseHandler,
+    );
+    unawaited(content);
+    yield* streamController.stream;
+  }
 }

lib/src/binding_http/http_client.dart

@@ -36,7 +36,8 @@ const _authorizationHeader = "Authorization";
 /// [RFC 7616]: https://datatracker.ietf.org/doc/html/rfc7616
 /// [RFC 6750]: https://datatracker.ietf.org/doc/html/rfc6750
 /// [`ComboSecurityScheme`]: https://w3c.github.io/wot-thing-description/#combosecurityscheme
-final class HttpClient extends ProtocolClient {
+final class HttpClient extends ProtocolClient
+    with DirectDiscoverer, CoreLinkFormatDiscoverer {
   /// Creates a new [HttpClient].
   HttpClient({
     AsyncClientSecurityCallback<BasicCredentials>? basicCredentialsCallback,
@@ -304,13 +305,13 @@ final class HttpClient extends ProtocolClient {
   }
 
   @override
-  Stream<DiscoveryContent> discoverDirectly(
+  Future<DiscoveryContent> discoverDirectly(
     Uri uri, {
     bool disableMulticast = false,
-  }) async* {
+  }) async {
     final request = Request(HttpRequestMethod.get.methodName, uri);
 
-    yield await _sendDiscoveryRequest(
+    return _sendDiscoveryRequest(
       request,
       acceptHeaderValue: "application/td+json",
     );
@@ -327,18 +328,4 @@ final class HttpClient extends ProtocolClient {
 
     yield encodedLinks;
   }
-
-  @override
-  Future<Content> requestThingDescription(Uri url) async {
-    final request = Request(HttpRequestMethod.get.methodName, url);
-    const tdContentType = "application/td+json";
-    request.headers["Accept"] = tdContentType;
-
-    final response = await _client.send(request);
-
-    return Content(
-      response.headers["Content-Type"] ?? tdContentType,
-      response.stream,
-    );
-  }
 }

lib/src/binding_mqtt/mqtt_client.dart

@@ -197,62 +197,4 @@ final class MqttClient extends ProtocolClient {
 
     return MqttSubscription(form, client, complete, next: next, error: error);
   }
-
-  @override
-  Stream<DiscoveryContent> discoverDirectly(
-    Uri uri, {
-    bool disableMulticast = false,
-  }) async* {
-    final client = await _connect(uri, null);
-    const discoveryTopic = "wot/td/#";
-
-    final streamController = StreamController<DiscoveryContent>();
-
-    Timer(
-      _mqttConfig.discoveryTimeout,
-      () async {
-        client.disconnect();
-        await streamController.close();
-      },
-    );
-
-    // TODO: Revisit QoS value and subscription check
-    if (client.subscribe(discoveryTopic, MqttQos.atLeastOnce) == null) {
-      throw MqttBindingException(
-        "Subscription to topic $discoveryTopic failed",
-      );
-    }
-
-    client.updates?.listen(
-      (messages) {
-        for (final message in messages) {
-          final publishedMessage = message.payload as MqttPublishMessage;
-          final payload = publishedMessage.payload.message;
-
-          streamController.add(
-            DiscoveryContent(
-              discoveryContentType,
-              Stream.value(payload),
-              uri,
-            ),
-          );
-        }
-      },
-      cancelOnError: false,
-    );
-
-    yield* streamController.stream;
-  }
-
-  @override
-  Stream<DiscoveryContent> discoverWithCoreLinkFormat(Uri uri) {
-    // TODO: implement discoverWithCoreLinkFormat
-    throw UnimplementedError();
-  }
-
-  @override
-  Future<Content> requestThingDescription(Uri url) {
-    // TODO: implement requestThingDescription
-    throw UnimplementedError();
-  }
 }

lib/src/core/implementation/servient.dart

@@ -280,9 +280,16 @@ class InternalServient implements Servient {
 
   /// Requests a [ThingDescription] from a [url].
   Future<ThingDescription> requestThingDescription(Uri url) async {
-    final client = clientFor(url.scheme);
-    final content = await client.requestThingDescription(url);
+    final uriScheme = url.scheme;
+    final client = clientFor(uriScheme);
 
+    if (client is! DirectDiscoverer) {
+      throw DiscoveryException(
+        "Client with URI scheme $uriScheme does not support direct discovery.",
+      );
+    }
+
+    final content = await client.discoverDirectly(url);
     final dataSchemaValue = await contentSerdes.contentToValue(content, null);
 
     if (dataSchemaValue

lib/src/core/implementation/thing_discovery.dart

@@ -12,6 +12,7 @@ import "package:multicast_dns/multicast_dns.dart";
 
 import "../definitions.dart";
 import "../exceptions.dart";
+import "../extensions.dart";
 import "../protocol_interfaces.dart";
 import "../scripting_api.dart" as scripting_api;
 
@@ -63,7 +64,11 @@ class ThingDiscovery extends Stream<ThingDescription>
           ):
           yield* _discoverFromCoreResourceDirectory(uri, discoveryType);
         case DirectConfiguration(:final uri):
-          yield* Stream.fromFuture(_servient.requestThingDescription(uri));
+          if (!uri.hasMulticastAddress) {
+            yield* Stream.fromFuture(_servient.requestThingDescription(uri));
+          } else {
+            yield* _performMulticastDiscovery(uri);
+          }
         case ExploreDirectoryConfiguration(:final uri, :final thingFilter):
           final thingDiscoveryProcess = await _servient.exploreDirectory(
             uri,
@@ -190,7 +195,18 @@ class ThingDiscovery extends Stream<ThingDescription>
     Uri uri,
     String resourceType,
   ) async* {
-    final client = _clientForUriScheme(uri.scheme);
+    final uriScheme = uri.scheme;
+    final client = _clientForUriScheme(uriScheme);
+
+    if (client is! CoreLinkFormatDiscoverer) {
+      yield* Stream.error(
+        DiscoveryException(
+          "Client for URI scheme $uriScheme does not support Core Link Format "
+          "Discovery.",
+        ),
+      );
+      return;
+    }
 
     await for (final coreWebLink in client.discoverWithCoreLinkFormat(uri)) {
       try {
@@ -251,7 +267,7 @@ class ThingDiscovery extends Stream<ThingDescription>
 
     if (dataSchemaValue is! scripting_api.DataSchemaValue<String>) {
       throw DiscoveryException(
-        "Could not parse Thing Description obtained from $sourceUri",
+        "Could not parse CoRE web links obtained from $sourceUri",
       );
     }
 
@@ -319,6 +335,45 @@ class ThingDiscovery extends Stream<ThingDescription>
 
     return Map.fromEntries(recordsList);
   }
+
+  Stream<ThingDescription> _performMulticastDiscovery(Uri uri) async* {
+    final client = _clientForUriScheme(uri.scheme);
+
+    if (client is MulticastDiscoverer) {
+      final contentStream = client.discoverViaMulticast(uri);
+      yield* _transformContentStreamToThingDescriptions(contentStream);
+    }
+  }
+
+  Stream<ThingDescription> _transformContentStreamToThingDescriptions(
+    Stream<Content> contentStream,
+  ) async* {
+    await for (final content in contentStream) {
+      try {
+        final thingDescription =
+            await _convertContentToThingDescription(content);
+        yield thingDescription;
+      } on Exception catch (exception) {
+        yield* Stream.error(exception);
+      }
+    }
+  }
+
+  Future<ThingDescription> _convertContentToThingDescription(
+    Content content,
+  ) async {
+    final dataSchemaValue =
+        await _servient.contentSerdes.contentToValue(content, null);
+
+    if (dataSchemaValue is scripting_api.ObjectValue) {
+      return dataSchemaValue.value.toThingDescription();
+    }
+
+    throw ValidationException(
+      "Encountered wrong datatype ${dataSchemaValue.runtimeType} that cannot "
+      "be processed as a Thing Description.",
+    );
+  }
 }
 
 extension _CoreLinkFormatExtension on String {

lib/src/core/protocol_interfaces.dart

@@ -6,5 +6,6 @@
 
 export "protocol_interfaces/protocol_client.dart";
 export "protocol_interfaces/protocol_client_factory.dart";
+export "protocol_interfaces/protocol_discoverer.dart";
 export "protocol_interfaces/protocol_server.dart";
 export "protocol_interfaces/protocol_subscription.dart";

lib/src/core/protocol_interfaces/protocol_client.dart

@@ -15,33 +15,6 @@ abstract base class ProtocolClient {
   /// Stops this [ProtocolClient].
   Future<void> stop();
 
-  /// Discovers one or more Thing Descriptions from a [uri], returning a
-  /// [Stream] of [Content].
-  ///
-  /// Allows the caller to explicitly [disableMulticast], overriding the
-  /// multicast settings in the config of the underlying binding implementation.
-  Stream<DiscoveryContent> discoverDirectly(
-    Uri uri, {
-    bool disableMulticast = false,
-  });
-
-  /// Discovers links using the CoRE Link Format (see [RFC 6690]) from a [uri],
-  /// encoded as a [Stream] of [Content].
-  ///
-  /// This method will also be used for discovery from CoRE Resource
-  /// Directories ([RFC 9176]).
-  ///
-  /// If the [uri]'s path is empty, then `/.well-known/core` will be set as a
-  /// default value.
-  ///
-  /// Certain protocols (like CoAP) might also use multicast for this discovery
-  /// method if the underlying binding implementation supports it and if it is
-  /// activated in the config.
-  ///
-  /// [RFC 6690]: https://datatracker.ietf.org/doc/html/rfc6690
-  /// [RFC 9176]: https://datatracker.ietf.org/doc/html/rfc9176
-  Stream<DiscoveryContent> discoverWithCoreLinkFormat(Uri uri);
-
   /// Requests the client to perform a `readproperty` operation on a [form].
   Future<Content> readResource(AugmentedForm form);
 
@@ -61,7 +34,4 @@ abstract base class ProtocolClient {
     void Function(Exception error)? error,
     required void Function() complete,
   });
-
-  /// Requests a Thing Description as [Content] from a [url].
-  Future<Content> requestThingDescription(Uri url);
 }