minor adjustments

Author: hiroshihorie

lib/src/token_source/caching.dart

@@ -14,6 +14,7 @@
 
 import 'dart:async';
 
+import '../support/completer_manager.dart';
 import 'jwt.dart';
 import 'token_source.dart';
 
@@ -23,11 +24,11 @@ import 'token_source.dart';
 /// return `true` if the cached credentials are still valid for the given request.
 ///
 /// The default validator checks JWT expiration using [isResponseExpired].
-typedef TokenValidator = bool Function(TokenRequestOptions? options, TokenSourceResponse response);
+typedef TokenValidator = bool Function(TokenRequestOptions options, TokenSourceResponse response);
 
 /// A tuple containing the request options and response that were cached.
 class TokenStoreItem {
-  final TokenRequestOptions? options;
+  final TokenRequestOptions options;
   final TokenSourceResponse response;
 
   const TokenStoreItem({
@@ -44,7 +45,7 @@ abstract class TokenStore {
   /// Store credentials in the store.
   ///
   /// This replaces any existing cached credentials with the new ones.
-  Future<void> store(TokenRequestOptions? options, TokenSourceResponse response);
+  Future<void> store(TokenRequestOptions options, TokenSourceResponse response);
 
   /// Retrieve the cached credentials.
   ///
@@ -63,7 +64,7 @@ class InMemoryTokenStore implements TokenStore {
   TokenStoreItem? _cached;
 
   @override
-  Future<void> store(TokenRequestOptions? options, TokenSourceResponse response) async {
+  Future<void> store(TokenRequestOptions options, TokenSourceResponse response) async {
     _cached = TokenStoreItem(options: options, response: response);
   }
 
@@ -79,7 +80,7 @@ class InMemoryTokenStore implements TokenStore {
 }
 
 /// Default validator that checks JWT expiration using [isResponseExpired].
-bool _defaultValidator(TokenRequestOptions? options, TokenSourceResponse response) {
+bool _defaultValidator(TokenRequestOptions options, TokenSourceResponse response) {
   return !isResponseExpired(response);
 }
 
@@ -96,7 +97,7 @@ class CachingTokenSource implements TokenSourceConfigurable {
   final TokenSourceConfigurable _wrapped;
   final TokenStore _store;
   final TokenValidator _validator;
-  Completer<TokenSourceResponse>? _fetchInProgress;
+  final Map<TokenRequestOptions, CompleterManager<TokenSourceResponse>> _inflightRequests = {};
 
   /// Initialize a caching wrapper around any token source.
   ///
@@ -112,31 +113,32 @@ class CachingTokenSource implements TokenSourceConfigurable {
         _validator = validator ?? _defaultValidator;
 
   @override
-  Future<TokenSourceResponse> fetch([TokenRequestOptions? options]) async {
-    if (_fetchInProgress != null) {
-      return _fetchInProgress!.future;
+  Future<TokenSourceResponse> fetch(TokenRequestOptions options) async {
+    final existingManager = _inflightRequests[options];
+    if (existingManager != null && existingManager.isActive) {
+      return existingManager.future;
     }
 
-    _fetchInProgress = Completer<TokenSourceResponse>();
+    final manager = existingManager ?? CompleterManager<TokenSourceResponse>();
+    _inflightRequests[options] = manager;
+    final resultFuture = manager.future;
 
     try {
-      // Check if we have a valid cached token
       final cached = await _store.retrieve();
       if (cached != null && cached.options == options && _validator(cached.options, cached.response)) {
-        _fetchInProgress!.complete(cached.response);
-        return cached.response;
+        manager.complete(cached.response);
+        return resultFuture;
       }
 
-      final requestOptions = options ?? const TokenRequestOptions();
-      final response = await _wrapped.fetch(requestOptions);
+      final response = await _wrapped.fetch(options);
       await _store.store(options, response);
-      _fetchInProgress!.complete(response);
-      return response;
-    } catch (e) {
-      _fetchInProgress!.completeError(e);
+      manager.complete(response);
+      return resultFuture;
+    } catch (e, stackTrace) {
+      manager.completeError(e, stackTrace);
       rethrow;
     } finally {
-      _fetchInProgress = null;
+      _inflightRequests.remove(options);
     }
   }
 

lib/src/token_source/endpoint.dart

@@ -25,7 +25,7 @@ import 'token_source.dart';
 /// - Encodes the request parameters as [TokenRequestOptions] JSON in the request body
 /// - Includes any custom headers specified via [headers]
 /// - Expects the response to be decoded as [TokenSourceResponse] JSON
-/// - Validates HTTP status codes (200) and throws appropriate errors for failures
+/// - Validates HTTP status codes (200-299) and throws appropriate errors for failures
 class EndpointTokenSource implements TokenSourceConfigurable {
   /// The URL endpoint for token generation.
   /// This should point to your backend service that generates LiveKit tokens.
@@ -54,21 +54,31 @@ class EndpointTokenSource implements TokenSourceConfigurable {
   });
 
   @override
-  Future<TokenSourceResponse> fetch([TokenRequestOptions? options]) async {
-    final requestOptions = options ?? const TokenRequestOptions();
-    final requestBody = jsonEncode(requestOptions.toRequest().toJson());
+  Future<TokenSourceResponse> fetch(TokenRequestOptions options) async {
+    final requestBody = jsonEncode(options.toRequest().toJson());
     final uri = Uri.parse(url);
     final requestHeaders = {
       'Content-Type': 'application/json',
       ...headers,
     };
 
     final httpClient = client ?? http.Client();
-    final response = method.toUpperCase() == 'GET'
-        ? await httpClient.get(uri, headers: requestHeaders)
-        : await httpClient.post(uri, headers: requestHeaders, body: requestBody);
+    final shouldCloseClient = client == null;
+    late final http.Response response;
 
-    if (response.statusCode != 200) {
+    try {
+      final request = http.Request(method, uri);
+      request.headers.addAll(requestHeaders);
+      request.body = requestBody;
+      final streamedResponse = await httpClient.send(request);
+      response = await http.Response.fromStream(streamedResponse);
+    } finally {
+      if (shouldCloseClient) {
+        httpClient.close();
+      }
+    }
+
+    if (response.statusCode < 200 || response.statusCode >= 300) {
       throw Exception('Error generating token from endpoint $url: received ${response.statusCode} / ${response.body}');
     }
 

lib/src/token_source/token_source.dart

@@ -175,5 +175,5 @@ abstract class TokenSourceFixed {
 /// - [EndpointTokenSource]: For custom backend endpoints using LiveKit's JSON format
 /// - [CachingTokenSource]: For caching credentials (or use the `.cached()` extension method)
 abstract class TokenSourceConfigurable {
-  Future<TokenSourceResponse> fetch([TokenRequestOptions? options]);
+  Future<TokenSourceResponse> fetch(TokenRequestOptions options);
 }

test/token/caching_token_source_test.dart

@@ -34,11 +34,11 @@ void main() {
       final cachingSource = CachingTokenSource(mockSource);
 
       // First fetch
-      final result1 = await cachingSource.fetch();
+      final result1 = await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 1);
 
       // Second fetch should use cache
-      final result2 = await cachingSource.fetch();
+      final result2 = await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 1);
       expect(result2.participantToken, result1.participantToken);
     });
@@ -57,11 +57,11 @@ void main() {
       final cachingSource = CachingTokenSource(mockSource);
 
       // First fetch with expired token
-      await cachingSource.fetch();
+      await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 1);
 
       // Second fetch should refetch due to expiration
-      await cachingSource.fetch();
+      await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 2);
     });
 
@@ -190,14 +190,37 @@ void main() {
       final cachingSource = CachingTokenSource(mockSource);
 
       // Start multiple concurrent fetches
-      final futures = List.generate(5, (_) => cachingSource.fetch());
+      final futures = List.generate(5, (_) => cachingSource.fetch(const TokenRequestOptions()));
       final results = await Future.wait(futures);
 
       // Should only fetch once despite concurrent requests
       expect(fetchCount, 1);
       expect(results.every((r) => r.participantToken == results.first.participantToken), isTrue);
     });
 
+    test('concurrent fetches with different options fetch independently', () async {
+      var fetchCount = 0;
+      final mockSource = _MockTokenSource((options) async {
+        fetchCount++;
+        await Future.delayed(Duration(milliseconds: 50));
+        final token = _generateValidToken();
+        return TokenSourceResponse(
+          serverUrl: 'https://test.livekit.io',
+          participantToken: '$token-${options.roomName}',
+        );
+      });
+
+      final cachingSource = CachingTokenSource(mockSource);
+
+      final futureOne = cachingSource.fetch(const TokenRequestOptions(roomName: 'room-a'));
+      final futureTwo = cachingSource.fetch(const TokenRequestOptions(roomName: 'room-b'));
+
+      final responses = await Future.wait([futureOne, futureTwo]);
+
+      expect(fetchCount, 2);
+      expect(responses[0].participantToken == responses[1].participantToken, isFalse);
+    });
+
     test('invalidate clears cache', () async {
       var fetchCount = 0;
       final mockSource = _MockTokenSource((options) async {
@@ -211,14 +234,14 @@ void main() {
 
       final cachingSource = CachingTokenSource(mockSource);
 
-      await cachingSource.fetch();
+      await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 1);
 
       // Invalidate cache
       await cachingSource.invalidate();
 
       // Should refetch after invalidation
-      await cachingSource.fetch();
+      await cachingSource.fetch(const TokenRequestOptions());
       expect(fetchCount, 2);
     });
 
@@ -235,7 +258,7 @@ void main() {
 
       expect(await cachingSource.cachedResponse(), isNull);
 
-      final response = await cachingSource.fetch();
+      final response = await cachingSource.fetch(const TokenRequestOptions());
       expect(await cachingSource.cachedResponse(), isNotNull);
       expect((await cachingSource.cachedResponse())?.participantToken, response.participantToken);
 

test/token/endpoint_token_source_test.dart

@@ -84,7 +84,7 @@ void main() {
       expect(agents[0]['metadata'], 'agent-metadata');
     });
 
-    test('GET endpoint', () async {
+    test('GET endpoint with body', () async {
       http.Request? capturedRequest;
 
       final mockClient = MockClient((request) async {
@@ -104,13 +104,37 @@ void main() {
         client: mockClient,
       );
 
-      final response = await source.fetch();
+      final response = await source.fetch(const TokenRequestOptions());
 
       expect(response.serverUrl, 'wss://www.example.com');
       expect(response.participantToken, 'token');
 
       expect(capturedRequest, isNotNull);
       expect(capturedRequest!.method, 'GET');
+      // Body is always sent even for GET requests
+      expect(capturedRequest!.body, '{}');
+    });
+
+    test('accepts non-200 success responses', () async {
+      final mockClient = MockClient((request) async {
+        return http.Response(
+          jsonEncode({
+            'server_url': 'wss://www.example.com',
+            'participant_token': 'token',
+          }),
+          201,
+        );
+      });
+
+      final source = EndpointTokenSource(
+        url: 'https://example.com/token',
+        client: mockClient,
+      );
+
+      final response = await source.fetch(const TokenRequestOptions());
+
+      expect(response.serverUrl, 'wss://www.example.com');
+      expect(response.participantToken, 'token');
     });
 
     test('camelCase backward compatibility', () async {
@@ -131,7 +155,7 @@ void main() {
         client: mockClient,
       );
 
-      final response = await source.fetch();
+      final response = await source.fetch(const TokenRequestOptions());
 
       expect(response.serverUrl, 'wss://www.example.com');
       expect(response.participantToken, 'token');
@@ -155,7 +179,7 @@ void main() {
         client: mockClient,
       );
 
-      final response = await source.fetch();
+      final response = await source.fetch(const TokenRequestOptions());
 
       expect(response.serverUrl, 'wss://www.example.com');
       expect(response.participantToken, 'token');
@@ -174,7 +198,7 @@ void main() {
       );
 
       expect(
-        () => source.fetch(),
+        () => source.fetch(const TokenRequestOptions()),
         throwsA(isA<Exception>().having(
           (e) => e.toString(),
           'message',
@@ -194,7 +218,7 @@ void main() {
       );
 
       expect(
-        () => source.fetch(),
+        () => source.fetch(const TokenRequestOptions()),
         throwsA(isA<Exception>().having(
           (e) => e.toString(),
           'message',

test/token/token_source_test.dart

@@ -160,7 +160,7 @@ void main() {
       }
 
       final source = CustomTokenSource(customFunction);
-      final result = await source.fetch();
+      final result = await source.fetch(const TokenRequestOptions());
 
       expect(result.serverUrl, 'https://custom.livekit.io');
       expect(result.participantToken, 'custom-token');