Support grpc-web in pure dart

lib/src/client/transport/xhr_transport.dart

@@ -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 HttpRequest _request;
+  final Client _client;
+  final Request _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,53 +41,34 @@ class XhrTransportStream implements GrpcTransportStream {
   @override
   StreamSink<List<int>> get outgoingMessages => _outgoingMessages.sink;
 
-  XhrTransportStream(this._request, {onError, onDone})
+  XhrTransportStream(this._client, this._request, {onError, onDone})
       : _onError = onError,
         _onDone = onDone {
-    _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) {
+    final asyncOnError = (e, st) {
       if (_incomingMessages.isClosed) {
         return;
       }
       _onError(GrpcError.unavailable('XhrConnection connection-error'));
       terminate();
-    });
-
-    _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);
-    });
+    };
+    _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);
 
     _incomingProcessor.stream
         .transform(GrpcWebDecoder())
@@ -96,30 +77,27 @@ class XhrTransportStream implements GrpcTransportStream {
             onError: _onError, onDone: _incomingMessages.close);
   }
 
-  _onHeadersReceived() {
-    final contentType = _request.getResponseHeader('Content-Type');
-    if (_request.status != 200) {
+  bool _onHeadersReceived(StreamedResponse response) {
+    final contentType = response.headers['content-type'];
+    if (response.statusCode != 200) {
       _onError(
-          GrpcError.unavailable('XhrConnection status ${_request.status}'));
-      return;
+          GrpcError.unavailable('XhrConnection status ${response.statusCode}'));
+      return false;
     }
     if (contentType == null) {
       _onError(GrpcError.unavailable('XhrConnection missing Content-Type'));
-      return;
+      return false;
     }
     if (!contentType.startsWith('application/grpc')) {
       _onError(
           GrpcError.unavailable('XhrConnection bad Content-Type $contentType'));
-      return;
-    }
-    if (_request.response == null) {
-      _onError(GrpcError.unavailable('XhrConnection request null response'));
-      return;
+      return false;
     }
 
     // Force a metadata message with headers.
-    final headers = GrpcMetadata(_request.responseHeaders);
+    final headers = GrpcMetadata(response.headers);
     _incomingMessages.add(headers);
+    return true;
   }
 
   _close() {
@@ -131,45 +109,47 @@ 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);
+  XhrClientConnection(this.uri) {
+    _client = createClient();
+  }
 
   String get authority => uri.authority;
   String get scheme => uri.scheme;
 
-  void _initializeRequest(HttpRequest request, Map<String, String> metadata) {
+  void _initializeRequest(Request request, Map<String, String> metadata) {
     for (final header in metadata.keys) {
-      request.setRequestHeader(header, metadata[header]);
+      request.headers[header] = metadata[header];
     }
-    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';
+    request.headers['Content-Type'] = 'application/grpc-web+proto';
+    request.headers['X-User-Agent'] = 'grpc-web-dart/0.1';
+    request.headers['X-Grpc-Web'] = '1';
   }
 
   @visibleForTesting
-  HttpRequest createHttpRequest() => HttpRequest();
+  Request createHttpRequest(String path) => Request('POST', uri.resolve(path));
+
+  @visibleForTesting
+  Client createClient() => Client();
 
   @override
   GrpcTransportStream makeRequest(String path, Duration timeout,
       Map<String, String> metadata, ErrorHandler onError) {
-    final HttpRequest request = createHttpRequest();
-    request.open('POST', uri.resolve(path).toString());
+    final Request request = createHttpRequest(path);
 
     _initializeRequest(request, metadata);
 
-    final XhrTransportStream transportStream =
-        XhrTransportStream(request, onError: onError, onDone: _removeStream);
+    final XhrTransportStream transportStream = XhrTransportStream(
+        _client, request,
+        onError: onError, onDone: _removeStream);
     _requests.add(transportStream);
     return transportStream;
   }