feat(kinesis): Kinesis Data Streams Library#6516
Conversation
- Add Auth: fetch session, explicit sign-out, sign-in with credentials - Add Storage: list files, delete files (cleanup) - Add API: query Todo by ID, delete Todo (cleanup) - Add DataStore: query with predicate, delete (cleanup) - Remove redundant Analytics guest tests - Add unique test data naming with UUIDs
…m/aws-amplify/amplify-flutter into feat/kinesis-data-streams-library
ekjotmultani
changed the title
| int maxBytes = 5 * 1024 * 1024, | ||
| }) async { | ||
| // Query records sorted by stream_name, partition_key, id | ||
| final query = _db.select(_db.kinesisRecords) |
There was a problem hiding this comment.
Can we optimize this query to avoid consuming high amounts of RAM by loading data we don't process? See also the Android implementation.
| retryableRecordIndices: List.generate(records.length, (i) => i), | ||
| ); | ||
| } on SmithyHttpException catch (e) { | ||
| // Check if it's a retryable HTTP error (5xx) |
There was a problem hiding this comment.
Good catch, addressing
| /// {@endtemplate} | ||
| class NetworkException extends KinesisDataStreamsException { | ||
| /// {@macro aws_kinesis_datastreams.network_exception} | ||
| const NetworkException(super.message, {super.underlyingException}) |
There was a problem hiding this comment.
How are we using this exception actually?
There was a problem hiding this comment.
For now it serves as being here while vNext common error classes are coming, should be thrown when network errors arise in send operations via the kinesis client, will address in commit
| final Future<void> Function() _onFlush; | ||
| Timer? _timer; | ||
| bool _enabled = true; | ||
| bool _closed = false; |
There was a problem hiding this comment.
What is the benefit of having both _enabled and _closed?
There was a problem hiding this comment.
_closed can serve as a flag for other code to know whether or not the client itself is shut down or if just flushing is enabled or not. In our implementation they do similar things but it is useful for getters to know the internal state of the client I feel. What do you think?
| final entry = resultEntries[i]; | ||
|
|
||
| if (entry.errorCode == null) { | ||
| // Success - record has sequenceNumber and shardId |
There was a problem hiding this comment.
record has sequenceNumber and shardId
Is this comment relevant? How are we using shardId?
There was a problem hiding this comment.
we arent using them, the comment just documents that we must have been returned these from the sdk call
| await client.flush(); | ||
|
|
||
| // Assert - if no exception thrown, the record was sent successfully | ||
| // The flush would throw if the stream doesn't exist or credentials are invalid |
There was a problem hiding this comment.
Can we have a more comprehensive assert here?
| } | ||
| await client.flush(); | ||
|
|
||
| // Assert - all records sent without error |
| ║ 2. Fill in your AWS credentials and stream name ║ | ||
| ║ 3. Set isConfigured = true ║ | ||
| ║ 4. Run: dart test test/e2e/ --tags=e2e ║ | ||
| ╚══════════════════════════════════════════════════════════════════╝ |
There was a problem hiding this comment.
Can we actually implement the e2e test before merging?
There was a problem hiding this comment.
yes before merge these tests will be live
| client.enable(); | ||
| await client.flush(); | ||
|
|
||
| // Assert - no exception means success (nothing was queued) |
| }); | ||
| }); | ||
| }); | ||
| } |
There was a problem hiding this comment.
Can we ensure concurrent access to the record storage works via a test?
…ass underlying exception for now
…s closed, added a unit test for this scenario
… e2e and unit tests where applicable
2 participants
Issue #, if available:
Description of changes:
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.