blob: 57048956480c573b6e768b1c4e040afb41f1604f [file] [log] [blame]
// Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
part of dds;
class _StreamManager {
_StreamManager(this.dds);
/// Send `streamNotify` notifications to clients subscribed to `streamId`.
///
/// If `data` is of type `Uint8List`, the notification is assumed to be a
/// binary event and is forwarded directly over the subscriber's websocket.
/// Otherwise, the event is sent via the JSON RPC client.
///
/// If `excludedClient` is provided, the notification will be sent to all
/// clients subscribed to `streamId` except for `excludedClient`.
void streamNotify(
String streamId,
data, {
_DartDevelopmentServiceClient excludedClient,
}) {
if (streamListeners.containsKey(streamId)) {
final listeners = streamListeners[streamId];
final isBinaryData = data is Uint8List;
for (final listener in listeners) {
if (listener == excludedClient) {
continue;
}
if (isBinaryData) {
listener.connection.sink.add(data);
} else {
listener.sendNotification('streamNotify', data);
}
}
}
}
void sendServiceRegisteredEvent(
_DartDevelopmentServiceClient client,
String service,
String alias,
) {
final namespace = dds._getNamespace(client);
streamNotify(
kServiceStream,
{
'streamId': kServiceStream,
'event': {
'type': 'Event',
'kind': 'ServiceRegistered',
'timestamp': DateTime.now().millisecondsSinceEpoch,
'service': service,
'method': namespace + '.' + service,
'alias': alias,
},
},
excludedClient: client,
);
}
void _sendServiceUnregisteredEvents(
_DartDevelopmentServiceClient client,
) {
final namespace = dds._getNamespace(client);
for (final service in client.services.keys) {
streamNotify(
kServiceStream,
{
'streamId': kServiceStream,
'event': {
'type': 'Event',
'kind': 'ServiceUnregistered',
'timestamp': DateTime.now().millisecondsSinceEpoch,
'service': service,
'method': namespace + '.' + service,
},
},
excludedClient: client,
);
}
}
/// Start listening for `streamNotify` events from the VM service and forward
/// them to the clients which have subscribed to the stream.
Future<void> listen() async {
// The _IsolateManager requires information from both the Debug and
// Isolate streams, so they must always be subscribed to by DDS.
for (final stream in ddsCoreStreams) {
await streamListen(null, stream);
}
dds._vmServiceClient.registerMethod(
'streamNotify',
(parameters) {
final streamId = parameters['streamId'].asString;
// Forward events from the streams _IsolateManager subscribes to.
if (isolateManagerStreams.contains(streamId)) {
dds.isolateManager.handleIsolateEvent(parameters);
}
// Keep a history of log messages to send to clients when they first
// subscribe to the Logging stream.
if (streamId == kLoggingStream) {
dds.loggingRepository.add(parameters.asMap);
}
streamNotify(streamId, parameters.value);
},
);
}
/// Subscribes `client` to a stream.
///
/// If `client` is the first client to listen to `stream`, DDS will send a
/// `streamListen` request for `stream` to the VM service.
Future<void> streamListen(
_DartDevelopmentServiceClient client,
String stream,
) async {
assert(stream != null && stream.isNotEmpty);
if (!streamListeners.containsKey(stream)) {
if ((stream == kDebugStream && client == null) ||
stream != kDebugStream) {
// This will return an RPC exception if the stream doesn't exist. This
// will throw and the exception will be forwarded to the client.
final result = await dds._vmServiceClient.sendRequest('streamListen', {
'streamId': stream,
});
assert(result['type'] == 'Success');
}
streamListeners[stream] = <_DartDevelopmentServiceClient>[];
}
if (streamListeners[stream].contains(client)) {
throw kStreamAlreadySubscribedException;
}
if (client != null) {
streamListeners[stream].add(client);
if (stream == kLoggingStream) {
dds.loggingRepository.sendHistoricalLogs(client);
}
}
}
/// Unsubscribes `client` from a stream.
///
/// If `client` is the last client to unsubscribe from `stream`, DDS will
/// send a `streamCancel` request for `stream` to the VM service.
Future<void> streamCancel(
_DartDevelopmentServiceClient client,
String stream, {
bool cancelCoreStream = false,
}) async {
assert(stream != null && stream.isNotEmpty);
final listeners = streamListeners[stream];
if (client != null && (listeners == null || !listeners.contains(client))) {
throw kStreamNotSubscribedException;
}
listeners.remove(client);
// Don't cancel streams DDS needs to function.
if (listeners.isEmpty &&
(!ddsCoreStreams.contains(stream) || cancelCoreStream)) {
streamListeners.remove(stream);
// Ensure the VM service hasn't shutdown.
if (dds._vmServiceClient.isClosed) {
return;
}
final result = await dds._vmServiceClient.sendRequest('streamCancel', {
'streamId': stream,
});
assert(result['type'] == 'Success');
} else {
streamListeners[stream] = listeners;
}
}
/// Cleanup stream subscriptions for `client` when it has disconnected.
void clientDisconnect(_DartDevelopmentServiceClient client) {
for (final streamId in streamListeners.keys.toList()) {
streamCancel(client, streamId).catchError(
(_) => null,
// Ignore 'stream not subscribed' errors and StateErrors which arise
// when DDS is shutting down.
test: (e) =>
(e is json_rpc.RpcException) ||
(dds._shuttingDown && e is StateError),
);
}
// Notify other service clients of service extensions that are being
// unregistered.
_sendServiceUnregisteredEvents(client);
}
static const kServiceStream = 'Service';
static final kStreamAlreadySubscribedException =
_RpcErrorCodes.buildRpcException(
_RpcErrorCodes.kStreamAlreadySubscribed,
);
static final kStreamNotSubscribedException = _RpcErrorCodes.buildRpcException(
_RpcErrorCodes.kStreamNotSubscribed,
);
static const kDebugStream = 'Debug';
static const kIsolateStream = 'Isolate';
static const kLoggingStream = 'Logging';
// Never cancel the Debug or Isolate stream as `_IsolateManager` requires
// them for isolate state notifications.
static const isolateManagerStreams = <String>{
kDebugStream,
kIsolateStream,
};
// Never cancel the Logging stream as `_LoggingRepository` requires it to
// keep a log history.
static const loggingRepositoryStreams = <String>{
kLoggingStream,
};
// The set of streams that DDS requires to function.
static final ddsCoreStreams = <String>{
...isolateManagerStreams,
...loggingRepositoryStreams,
};
final _DartDevelopmentService dds;
final streamListeners = <String, List<_DartDevelopmentServiceClient>>{};
}