grpc
/
grpc-dart
mirror of https://github.com/grpc/grpc-dart.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Revert "Support grpc-web in pure dart (#287)" (#351) 

This reverts commit c513e1467f.

The original commit has broken streaming due to limitations of package:http.
This commit is contained in:
Vyacheslav Egorov 2020-09-17 19:59:15 +02:00 committed by GitHub
parent 6fa4616bac
commit 3414356950
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 165 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
lib/src/client/transport/xhr_transport.dart
View File

@ -14,9 +14,9 @@
// limitations under the License.
import 'dart:async';
import 'dart:html';
import 'dart:typed_data';
import 'package:http/http.dart';
import 'package:meta/meta.dart';
import '../../client/call.dart';
@ -27,10 +27,10 @@ import 'transport.dart';
import 'web_streams.dart';
class XhrTransportStream implements GrpcTransportStream {
final Client _client;
final Request _request;
final HttpRequest _request;
final ErrorHandler _onError;
final Function(XhrTransportStream stream) _onDone;
int _requestBytesRead = 0;
final StreamController<ByteBuffer> _incomingProcessor = StreamController();
final StreamController<GrpcMessage> _incomingMessages = StreamController();
final StreamController<List<int>> _outgoingMessages = StreamController();
@ -41,34 +41,53 @@ class XhrTransportStream implements GrpcTransportStream {
@override
StreamSink<List<int>> get outgoingMessages => _outgoingMessages.sink;
XhrTransportStream(this._client, this._request, {onError, onDone})
XhrTransportStream(this._request, {onError, onDone})
: _onError = onError,
_onDone = onDone {
final asyncOnError = (e, st) {
_outgoingMessages.stream
.map(frame)
.listen((data) => _request.send(data), cancelOnError: true);
_request.onReadyStateChange.listen((data) {
if (_incomingMessages.isClosed) {
return;
}
switch (_request.readyState) {
case HttpRequest.HEADERS_RECEIVED:
_onHeadersReceived();
break;
case HttpRequest.DONE:
if (_request.status != 200) {
_onError(GrpcError.unavailable(
'XhrConnection status ${_request.status}'));
} else {
_close();
}
break;
}
});
_request.onError.listen((ProgressEvent event) {
if (_incomingMessages.isClosed) {
return;
}
_onError(GrpcError.unavailable('XhrConnection connection-error'));
terminate();
};
_outgoingMessages.stream.map(frame).listen((data) {
_request.bodyBytes = data;
var firstMessage = true;
_client.send(_request).then((response) {
if (_incomingMessages.isClosed) {
return;
}
if (firstMessage) {
if (!_onHeadersReceived(response)) {
return;
}
}
firstMessage = false;
response.stream.listen((data) {
_incomingProcessor.add(Uint8List.fromList(data).buffer);
}, onDone: _close);
}).catchError(asyncOnError);
}, cancelOnError: true, onError: asyncOnError);
});
_request.onProgress.listen((_) {
if (_incomingMessages.isClosed) {
return;
}
// Use response over responseText as most browsers don't support
// using responseText during an onProgress event.
final responseString = _request.response as String;
final bytes = Uint8List.fromList(
responseString.substring(_requestBytesRead).codeUnits)
.buffer;
_requestBytesRead = responseString.length;
_incomingProcessor.add(bytes);
});
_incomingProcessor.stream
.transform(GrpcWebDecoder())
@ -77,27 +96,30 @@ class XhrTransportStream implements GrpcTransportStream {
onError: _onError, onDone: _incomingMessages.close);
}
bool _onHeadersReceived(StreamedResponse response) {
final contentType = response.headers['content-type'];
if (response.statusCode != 200) {
_onHeadersReceived() {
final contentType = _request.getResponseHeader('Content-Type');
if (_request.status != 200) {
_onError(
GrpcError.unavailable('XhrConnection status ${response.statusCode}'));
return false;
GrpcError.unavailable('XhrConnection status ${_request.status}'));
return;
}
if (contentType == null) {
_onError(GrpcError.unavailable('XhrConnection missing Content-Type'));
return false;
return;
}
if (!contentType.startsWith('application/grpc')) {
_onError(
GrpcError.unavailable('XhrConnection bad Content-Type $contentType'));
return false;
return;
}
if (_request.response == null) {
_onError(GrpcError.unavailable('XhrConnection request null response'));
return;
}
// Force a metadata message with headers.
final headers = GrpcMetadata(response.headers);
final headers = GrpcMetadata(_request.responseHeaders);
_incomingMessages.add(headers);
return true;
}
_close() {
@ -109,47 +131,45 @@ class XhrTransportStream implements GrpcTransportStream {
@override
Future<void> terminate() async {
_close();
_request.abort();
}
}
class XhrClientConnection extends ClientConnection {
final Uri uri;
Client _client;
final Set<XhrTransportStream> _requests = Set<XhrTransportStream>();
XhrClientConnection(this.uri) {
_client = createClient();
}
XhrClientConnection(this.uri);
String get authority => uri.authority;
String get scheme => uri.scheme;
void _initializeRequest(Request request, Map<String, String> metadata) {
void _initializeRequest(HttpRequest request, Map<String, String> metadata) {
for (final header in metadata.keys) {
request.headers[header] = metadata[header];
request.setRequestHeader(header, metadata[header]);
}
request.headers['Content-Type'] = 'application/grpc-web+proto';
request.headers['X-User-Agent'] = 'grpc-web-dart/0.1';
request.headers['X-Grpc-Web'] = '1';
request.setRequestHeader('Content-Type', 'application/grpc-web+proto');
request.setRequestHeader('X-User-Agent', 'grpc-web-dart/0.1');
request.setRequestHeader('X-Grpc-Web', '1');
// Overriding the mimetype allows us to stream and parse the data
request.overrideMimeType('text/plain; charset=x-user-defined');
request.responseType = 'text';
}
@visibleForTesting
Request createHttpRequest(String path) => Request('POST', uri.resolve(path));
@visibleForTesting
Client createClient() => Client();
HttpRequest createHttpRequest() => HttpRequest();
@override
GrpcTransportStream makeRequest(String path, Duration timeout,
Map<String, String> metadata, ErrorHandler onError) {
final Request request = createHttpRequest(path);
final HttpRequest request = createHttpRequest();
request.open('POST', uri.resolve(path).toString());
_initializeRequest(request, metadata);
final XhrTransportStream transportStream = XhrTransportStream(
_client, request,
onError: onError, onDone: _removeStream);
final XhrTransportStream transportStream =
XhrTransportStream(request, onError: onError, onDone: _removeStream);
_requests.add(transportStream);
return transportStream;
}

187
test/client_tests/client_xhr_transport_test.dart
View File

@ -12,32 +12,49 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
@TestOn('browser')
import 'dart:async';
import 'dart:html';
import 'package:grpc/src/client/transport/xhr_transport.dart';
import 'package:grpc/src/shared/message.dart';
import 'package:http/http.dart';
import 'package:mockito/mockito.dart';
import 'package:test/test.dart';
class MockClient extends Mock implements Client {}
class MockHttpRequest extends Mock implements HttpRequest {
// ignore: close_sinks
StreamController<Event> readyStateChangeController =
StreamController<Event>();
// ignore: close_sinks
StreamController<ProgressEvent> progressController =
StreamController<ProgressEvent>();
class MockRequest extends Mock implements Request {}
@override
Stream<Event> get onReadyStateChange => readyStateChangeController.stream;
@override
Stream<ProgressEvent> get onProgress => progressController.stream;
@override
Stream<ProgressEvent> get onError => StreamController<ProgressEvent>().stream;
@override
int status = 200;
}
class MockXhrClientConnection extends XhrClientConnection {
MockXhrClientConnection() : super(Uri.parse('test:8080'));
MockRequest latestRequest = MockRequest();
final client = MockClient();
MockHttpRequest latestRequest;
@override
createHttpRequest(String path) {
return latestRequest;
}
@override
createClient() {
return client;
createHttpRequest() {
final request = MockHttpRequest();
latestRequest = request;
return request;
}
}
@ -49,16 +66,18 @@ void main() {
};
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
connection.makeRequest('path', Duration(seconds: 10), metadata,
(error) => fail(error.toString()));
expect(connection.latestRequest.headers['Content-Type'],
'application/grpc-web+proto');
expect(
connection.latestRequest.headers['X-User-Agent'], 'grpc-web-dart/0.1');
expect(connection.latestRequest.headers['X-Grpc-Web'], '1');
verify(connection.latestRequest
.setRequestHeader('Content-Type', 'application/grpc-web+proto'));
verify(connection.latestRequest
.setRequestHeader('X-User-Agent', 'grpc-web-dart/0.1'));
verify(connection.latestRequest.setRequestHeader('X-Grpc-Web', '1'));
verify(connection.latestRequest
.overrideMimeType('text/plain; charset=x-user-defined'));
verify(connection.latestRequest.responseType = 'text');
});
test('Sent data converted to stream properly', () async {
@ -68,41 +87,29 @@ void main() {
};
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
final stream = connection.makeRequest('path', Duration(seconds: 10),
metadata, (error) => fail(error.toString()));
when(connection.client.send(captureAny)).thenAnswer(
(_) => Future.value(StreamedResponse(Stream.fromIterable([]), 200)));
final data = List.filled(10, 0);
final expectedData = frame(data);
stream.outgoingMessages.add(data);
await stream.terminate();
verify(connection.latestRequest.bodyBytes = expectedData);
final expectedData = frame(data);
expect(verify(connection.latestRequest.send(captureAny)).captured.single,
expectedData);
});
test('Stream handles headers properly', () async {
final metadata = <String, String>{
'parameter_1': 'value_1',
'parameter_2': 'value_2',
'content-type': 'application/grpc+proto',
'parameter_2': 'value_2'
};
final transport = MockXhrClientConnection();
when(transport.latestRequest.headers).thenReturn({});
final stream = transport.makeRequest('test_path', Duration(seconds: 10),
metadata, (error) => fail(error.toString()));
when(transport.client.send(captureAny)).thenAnswer((_) {
return Future.value(
StreamedResponse(Stream.fromIterable([]), 200, headers: metadata));
});
final data = List.filled(10, 0);
stream.outgoingMessages.add(data);
stream.incomingMessages.listen((message) {
expect(message, TypeMatcher<GrpcMetadata>());
@ -117,76 +124,63 @@ void main() {
test('Stream handles trailers properly', () async {
final trailers = <String, String>{
'trailer_1': 'value_1',
'trailer_2': 'value_2',
'trailer_2': 'value_2'
};
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
final stream = connection.makeRequest('test_path', Duration(seconds: 10),
{}, (error) => fail(error.toString()));
final encodedTrailers = frame(trailers.entries
.map((e) => '${e.key}:${e.value}')
.join('\r\n')
.codeUnits);
encodedTrailers[0] = 0x80; // Mark this frame as trailers.
final response = StreamedResponse(
Future.value(encodedTrailers).asStream(), 200,
headers: {'content-type': 'application/grpc+proto'});
when(connection.client.send(connection.latestRequest))
.thenAnswer((_) => Future.value(response));
final encodedString = String.fromCharCodes(encodedTrailers);
final stream = connection.makeRequest('test_path', Duration(seconds: 10),
{}, (error) => fail(error.toString()));
final data = List.filled(10, 0);
stream.outgoingMessages.add(data);
bool first = true;
stream.incomingMessages.listen((message) {
expect(message, TypeMatcher<GrpcMetadata>());
if (message is GrpcMetadata) {
if (first) {
expect(message.metadata.length, 1);
expect(message.metadata.entries.first.key, 'content-type');
expect(
message.metadata.entries.first.value, 'application/grpc+proto');
first = false;
} else {
message.metadata.forEach((key, value) {
expect(value, trailers[key]);
});
}
message.metadata.forEach((key, value) {
expect(value, trailers[key]);
});
}
});
when(connection.latestRequest.getResponseHeader('Content-Type'))
.thenReturn('application/grpc+proto');
when(connection.latestRequest.responseHeaders).thenReturn({});
when(connection.latestRequest.readyState)
.thenReturn(HttpRequest.HEADERS_RECEIVED);
when(connection.latestRequest.response).thenReturn(encodedString);
connection.latestRequest.readyStateChangeController.add(null);
connection.latestRequest.progressController.add(null);
});
test('Stream handles empty trailers properly', () async {
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
final stream = connection.makeRequest('test_path', Duration(seconds: 10),
{}, (error) => fail(error.toString()));
final encoded = frame(''.codeUnits);
encoded[0] = 0x80; // Mark this frame as trailers.
final response = StreamedResponse(Future.value(encoded).asStream(), 200,
headers: {'content-type': 'application/grpc+proto'});
final encodedString = String.fromCharCodes(encoded);
when(connection.client.send(connection.latestRequest))
.thenAnswer((_) => Future.value(response));
final data = List.filled(10, 0);
stream.outgoingMessages.add(data);
bool trailer = false;
stream.incomingMessages.listen((message) {
expect(message, TypeMatcher<GrpcMetadata>());
if (message is GrpcMetadata) {
if (trailer) {
expect(message.metadata.isEmpty, true);
}
trailer = true;
message.metadata.isEmpty;
}
});
when(connection.latestRequest.getResponseHeader('Content-Type'))
.thenReturn('application/grpc+proto');
when(connection.latestRequest.responseHeaders).thenReturn({});
when(connection.latestRequest.readyState)
.thenReturn(HttpRequest.HEADERS_RECEIVED);
when(connection.latestRequest.response).thenReturn(encodedString);
connection.latestRequest.readyStateChangeController.add(null);
connection.latestRequest.progressController.add(null);
});
test('Stream deserializes data properly', () async {
@ -196,36 +190,36 @@ void main() {
};
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
final stream = connection.makeRequest('test_path', Duration(seconds: 10),
metadata, (error) => fail(error.toString()));
final data = List<int>.filled(10, 224);
final encoded = frame(data);
final response = StreamedResponse(Future.value(encoded).asStream(), 200,
headers: {'content-type': 'application/grpc+proto'});
stream.outgoingMessages.add(data);
when(connection.client.send(connection.latestRequest))
.thenAnswer((_) => Future.value(response));
final encodedString = String.fromCharCodes(encoded);
stream.incomingMessages.listen(expectAsync1((message) {
if (message is GrpcData) {
expect(message.data, equals(data));
}
}, count: 2));
when(connection.latestRequest.getResponseHeader('Content-Type'))
.thenReturn('application/grpc+proto');
when(connection.latestRequest.responseHeaders).thenReturn(metadata);
when(connection.latestRequest.readyState)
.thenReturn(HttpRequest.HEADERS_RECEIVED);
when(connection.latestRequest.response).thenReturn(encodedString);
connection.latestRequest.readyStateChangeController.add(null);
connection.latestRequest.progressController.add(null);
});
test('Stream recieves multiple messages', () async {
final metadata = <String, String>{
'parameter_1': 'value_1',
'parameter_2': 'value_2',
'content-type': 'application/grpc+proto',
'parameter_2': 'value_2'
};
final connection = MockXhrClientConnection();
when(connection.latestRequest.headers).thenReturn({});
final stream = connection.makeRequest('test_path', Duration(seconds: 10),
metadata, (error) => fail(error.toString()));
@ -235,14 +229,7 @@ void main() {
List<int>.filled(5, 124)
];
final encoded = data.map((d) => frame(d));
final response =
StreamedResponse(Stream.fromIterable(encoded), 200, headers: metadata);
when(connection.client.send(connection.latestRequest))
.thenAnswer((_) => Future.value(response));
final outData = List.filled(10, 0);
stream.outgoingMessages.add(outData);
final encodedStrings = encoded.map((e) => String.fromCharCodes(e)).toList();
final expectedMessages = <GrpcMessage>[
GrpcMetadata(metadata),
@ -260,5 +247,21 @@ void main() {
expect(message.data, (expectedMessage as GrpcData).data);
}
}, count: expectedMessages.length));
when(connection.latestRequest.getResponseHeader('Content-Type'))
.thenReturn('application/grpc+proto');
when(connection.latestRequest.responseHeaders).thenReturn(metadata);
when(connection.latestRequest.readyState)
.thenReturn(HttpRequest.HEADERS_RECEIVED);
// At first - expected response is the first message
when(connection.latestRequest.response)
.thenAnswer((_) => encodedStrings[0]);
connection.latestRequest.readyStateChangeController.add(null);
connection.latestRequest.progressController.add(null);
// After the first call, expected response should now be both responses together
when(connection.latestRequest.response)
.thenAnswer((_) => encodedStrings[0] + encodedStrings[1]);
connection.latestRequest.progressController.add(null);
});
}