-
Notifications
You must be signed in to change notification settings - Fork 0
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
fix: topics subscription resiliency #38
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,30 @@ | ||
import 'package:client_sdk_dart/client_sdk_dart.dart'; | ||
import 'package:client_sdk_dart/generated/cachepubsub.pbgrpc.dart'; | ||
import 'package:grpc/grpc.dart'; | ||
import 'package:logging/logging.dart'; | ||
|
||
class TopicGrpcManager { | ||
late final ClientChannel _channel; | ||
late final PubsubClient _client; | ||
final _logger = Logger('MomentoTopicClient'); | ||
|
||
TopicGrpcManager(CredentialProvider credentialProvider) { | ||
_channel = ClientChannel(credentialProvider.cacheEndpoint); | ||
_client = PubsubClient(_channel, | ||
options: CallOptions(metadata: { | ||
'authorization': credentialProvider.apiKey, | ||
'agent': 'dart:0.1.0' | ||
})); | ||
} | ||
|
||
PubsubClient get client => _client; | ||
ClientChannel get channel => _channel; | ||
|
||
void close() { | ||
try { | ||
_channel.shutdown(); | ||
} catch (e) { | ||
_logger.warning("Error shutting down channel: $e"); | ||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,24 +1,64 @@ | ||
import 'package:client_sdk_dart/generated/cachepubsub.pbgrpc.dart'; | ||
import 'package:client_sdk_dart/src/errors/errors.dart'; | ||
import 'package:client_sdk_dart/src/messages/responses/responses_base.dart'; | ||
import 'package:grpc/grpc.dart'; | ||
import 'package:logging/logging.dart'; | ||
import 'package:fixnum/fixnum.dart'; | ||
|
||
import '../../../internal/pubsub_client.dart'; | ||
import 'topic_subscription_item.dart'; | ||
|
||
sealed class TopicSubscribeResponse {} | ||
|
||
class TopicSubscription implements TopicSubscribeResponse { | ||
final ResponseStream<SubscriptionItem_> _stream; | ||
ResponseStream<SubscriptionItem_> _stream; | ||
Int64 lastSequenceNumber; | ||
final ClientPubsub _client; | ||
String cacheName; | ||
String topicName; | ||
bool retry = true; | ||
final logger = Logger("TopicSubscribeResponse"); | ||
|
||
TopicSubscription(this._stream); | ||
TopicSubscription(this._stream, this.lastSequenceNumber, this._client, | ||
this.cacheName, this.topicName); | ||
|
||
Stream<TopicSubscriptionItemResponse> get stream => | ||
_stream.map(_processResult).where((item) => item != null).cast(); | ||
Stream<TopicSubscriptionItemResponse> get stream { | ||
return _handleStream(); | ||
} | ||
|
||
Stream<TopicSubscriptionItemResponse> _handleStream() async* { | ||
while (retry) { | ||
try { | ||
await for (final msg in _stream) { | ||
var result = _processResult(msg); | ||
if (result != null) { | ||
yield result; | ||
} | ||
} | ||
} catch (e) { | ||
if (e is CancelledException || | ||
(e is GrpcError && e.codeName == "CANCELLED")) { | ||
logger.fine("Subscription was cancelled, not reconnecting"); | ||
retry = false; | ||
} else { | ||
logger.fine("Attempting to reconnect after receiving error: $e"); | ||
var result = _client.subscribe(cacheName, topicName, | ||
resumeAtTopicSequenceNumber: lastSequenceNumber); | ||
if (result is TopicSubscription) { | ||
_stream = result._stream; | ||
lastSequenceNumber = result.lastSequenceNumber; | ||
} else if (result is TopicSubscribeError) { | ||
logger.fine("Error reconnecting: ${result.message}"); | ||
} | ||
} | ||
} | ||
} | ||
} | ||
|
||
TopicSubscriptionItemResponse? _processResult(SubscriptionItem_ item) { | ||
final logger = Logger("TopicSubscribeResponse"); | ||
switch (item.whichKind()) { | ||
case SubscriptionItem__Kind.item: | ||
lastSequenceNumber = item.item.topicSequenceNumber; | ||
return createTopicItemResponse(item.item); | ||
case SubscriptionItem__Kind.heartbeat: | ||
logger.fine("topic client received a heartbeat"); | ||
|
@@ -30,6 +70,10 @@ class TopicSubscription implements TopicSubscribeResponse { | |
} | ||
return null; | ||
} | ||
|
||
void unsubscribe() async { | ||
await _stream.cancel(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 👍 |
||
} | ||
} | ||
|
||
class TopicSubscribeError extends ErrorResponseBase | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,5 +1,5 @@ | ||
import 'package:client_sdk_dart/src/auth/credential_provider.dart'; | ||
import 'package:client_sdk_dart/src/config/logger.dart'; | ||
// import 'package:client_sdk_dart/src/config/logger.dart'; | ||
import 'package:logging/logging.dart'; | ||
import 'config/topic_configuration.dart'; | ||
import 'internal/pubsub_client.dart'; | ||
|
@@ -12,6 +12,8 @@ abstract class ITopicClient { | |
String cacheName, String topicName, Value value); | ||
|
||
TopicSubscribeResponse subscribe(String cacheName, String topicName); | ||
|
||
void close(); | ||
} | ||
|
||
class TopicClient implements ITopicClient { | ||
|
@@ -21,7 +23,7 @@ class TopicClient implements ITopicClient { | |
TopicClient( | ||
CredentialProvider credentialProvider, TopicConfiguration configuration) | ||
: _pubsubClient = ClientPubsub(credentialProvider, configuration) { | ||
_logger.level = determineLoggerLevel(configuration.logLevel); | ||
// _logger.level = determineLoggerLevel(configuration.logLevel); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. errant comments, maybe remove in a later PR. |
||
_logger.finest("initializing topic client"); | ||
} | ||
|
||
|
@@ -35,4 +37,9 @@ class TopicClient implements ITopicClient { | |
TopicSubscribeResponse subscribe(String cacheName, String topicName) { | ||
return _pubsubClient.subscribe(cacheName, topicName); | ||
} | ||
|
||
@override | ||
void close() { | ||
_pubsubClient.close(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. :thu |
||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this is really useful, I imagine it was great as almost a TDD approach to working on this feature. In the long run though, we probably want to keep these examples as simple as humanly possible, so we might want to move this to an "advanced topics example" or something.
Not in any way, shape, or form a blocker for this PR :) just calling it out as something to be aware of for future touching up on these examples/docs and stuff.